`dapkel.core`¶

The shared substrate every analysis is built on. Anything copied into two analysis modules belongs here instead — tests/test_api_surface.py fails if a function name is defined twice.

See Adding an analysis for the contract.

`core.io` — finding and sizing raw files¶

io ¶

Stage 0: locating and sizing the raw Kelpie '.bin' files.

Every analysis starts by picking the '.bin' files that belong to a measurement and working out how many frames each holds. Both were previously copied into four modules; they live here now so the '.bin' format is described in exactly one place.

frames_in_file ¶

frames_in_file(fp: str) -> int

Return how many frames a '.bin' file has room for, from its size.

Do not use this as the acquisition's frame count.

The relation between the frame count an acquisition was asked for and the size of the '.bin' it produces is not established. Every sample file measured is a whole power-of-two number of frames - 32 MiB (16 384 frames) and 16 MiB (8 192) - which is what the exe's DDR3 transfer size would give for a power-of-two nframes, but not what a 10 000-frame acquisition should produce. Either those files were taken at 16 384 and 8 192, or something rounds; a controlled sweep on the rig is what settles it.

Both this package's 'unpack' and the reference kelpie_data_ddr3.m read only the first nframes frames, so the acquisition setting is what every analysis takes as a required argument, and it is the number a rate must be normalised by. Use this only when the file size really is the only record left, and treat it as an upper bound.

Parameters:

Name	Type	Description	Default
`fp`	`str`	Path to the '.bin' file.	required

Returns:

Type	Description
`int`	Frame capacity of the file, `filesize // BYTES_PER_FRAME`.

Raises:

Type	Description
`ValueError`	Raised when the file size is not a whole number of frames.

Source code in src/dapkel/core/io.py

def frames_in_file(fp: str) -> int:
    """Return how many frames a '.bin' file has *room* for, from its size.

    !!! warning "Do not use this as the acquisition's frame count."
        The relation between the frame count an acquisition was *asked* for and
        the size of the '.bin' it produces is **not established**. Every sample
        file measured is a whole power-of-two number of frames - 32 MiB
        (16 384 frames) and 16 MiB (8 192) - which is what the exe's DDR3
        transfer size would give for a power-of-two ``nframes``, but not what a
        10 000-frame acquisition should produce. Either those files were taken
        at 16 384 and 8 192, or something rounds; a controlled sweep on the rig
        is what settles it.

        Both this package's 'unpack' and the reference ``kelpie_data_ddr3.m``
        read only the first ``nframes`` frames, so the acquisition setting is
        what every analysis takes as a required argument, and it is the number
        a rate must be normalised by. Use this only when the file size really
        is the only record left, and treat it as an upper bound.

    Parameters
    ----------
    fp : str
        Path to the '.bin' file.

    Returns
    -------
    int
        Frame capacity of the file, ``filesize // BYTES_PER_FRAME``.

    Raises
    ------
    ValueError
        Raised when the file size is not a whole number of frames.
    """
    size = os.path.getsize(fp)
    if size == 0 or size % BYTES_PER_FRAME != 0:
        raise ValueError(
            f"{os.path.basename(fp)} is {size} bytes, not a whole number of "
            f"{BYTES_PER_FRAME}-byte frames."
        )
    return size // BYTES_PER_FRAME

find_bin_files ¶

find_bin_files(folder: str, tag: str, *, require_separator: bool = False) -> list[str]

Find and naturally sort the '.bin' files for a readout tag.

"Naturally" means ordered by the digits in the file name, so ..._2.bin sorts before ..._10.bin.

Parameters:

Name	Type	Description	Default
`folder`	`str`	Path to the folder with the '.bin' data files.	required
`tag`	`str`	Filename fragment selecting the files (e.g. `'ORT'`, `'S0C'`, `'SPAD0_S0'`), or `''` to match every '.bin' file in the folder.	required
`require_separator`	`bool`	When True the tag must be preceded by an underscore (pattern `'_<tag>.bin'` instead of `'<tag>.bin'`). The DCR quadrant tags need this so that, for example, `'S0C'` cannot also match a file whose name merely contains those characters. The default is False.	`False`

Returns:

Type	Description
`list[str]`	Naturally sorted list of paths to the matching '.bin' files.

Raises:

Type	Description
`FileNotFoundError`	Raised when no '.bin' files match the tag in the folder.

Source code in src/dapkel/core/io.py

def find_bin_files(
    folder: str, tag: str, *, require_separator: bool = False
) -> list[str]:
    """Find and naturally sort the '.bin' files for a readout tag.

    "Naturally" means ordered by the digits in the file name, so ``..._2.bin``
    sorts before ``..._10.bin``.

    Parameters
    ----------
    folder : str
        Path to the folder with the '.bin' data files.
    tag : str
        Filename fragment selecting the files (e.g. ``'ORT'``, ``'S0C'``,
        ``'SPAD0_S0'``), or ``''`` to match every '.bin' file in the folder.
    require_separator : bool, optional
        When True the tag must be preceded by an underscore (pattern
        ``'*_<tag>*.bin'`` instead of ``'*<tag>*.bin'``). The DCR quadrant
        tags need this so that, for example, ``'S0C'`` cannot also match a
        file whose name merely contains those characters. The default is
        False.

    Returns
    -------
    list[str]
        Naturally sorted list of paths to the matching '.bin' files.

    Raises
    ------
    FileNotFoundError
        Raised when no '.bin' files match the tag in the folder.
    """
    if tag:
        pattern = f"*_{tag}*.bin" if require_separator else f"*{tag}*.bin"
    else:
        pattern = "*.bin"

    files = sorted(
        glob.glob(os.path.join(folder, pattern)),
        key=lambda fp: int(
            "".join(filter(str.isdigit, os.path.basename(fp))) or "0"
        ),
    )
    if not files:
        raise FileNotFoundError(
            f"No .bin files matching '{pattern}' found in:\n  {folder}"
        )
    return files

`core.timing` — frame period and live time¶

timing ¶

Stage 1: resolving how much wall-clock time an acquisition represents.

Counts become rates only once you know how long the sensor was actually collecting, and that depends on the firmware. These helpers previously lived in 'dcr_analysis' as private functions that 'hitmap_analysis' imported across the module boundary; they are shared API, so they live here.

Also the single home for the ORT timing scale. Every time parameter in this package is in seconds and named with an _s suffix; picoseconds appear only as internal working units, because that is the scale the TDC lives on. Convert at the boundary with 'to_ps' / 'to_s'.

Two timescales are easy to conflate and must not be:

open_shutter_time_s - the photon-sensitive window per frame. Sets the rate normalisation, and bounds how far apart two detections in one frame can be.
the TDC full scale - the ORT measurement range, i.e. one period of the reference clock. The timestamp is a code on this range, so it is what the accidental (triangle) background's half-base follows.

They coincide only when the shutter is one clock period. When the shutter is longer, the free-running oscillator has no cycle counter, so timestamps fold into the full scale: a 500 ns shutter read by a 10 MHz (100 ns) TDC gives a 100 ns triangle, not a 500 ns one. Never assume one from the other - 'resolve_tdc_full_scale_s' takes the full scale from the calibration LUT, which is self-describing, and only falls back to the shutter when there is no LUT.

to_ps ¶

to_ps(seconds: float) -> float

Convert seconds to picoseconds.

Parameters:

Name	Type	Description	Default
`seconds`	`float`	A time in seconds (the package-wide parameter convention).	required

Returns:

Type	Description
`float`	The same time in picoseconds (the internal working unit).

Source code in src/dapkel/core/timing.py

def to_ps(seconds: float) -> float:
    """Convert seconds to picoseconds.

    Parameters
    ----------
    seconds : float
        A time in seconds (the package-wide parameter convention).

    Returns
    -------
    float
        The same time in picoseconds (the internal working unit).
    """
    return float(seconds) * 1e12

to_s ¶

to_s(picoseconds: float) -> float

Convert picoseconds to seconds.

Parameters:

Name	Type	Description	Default
`picoseconds`	`float`	A time in picoseconds (the internal working unit).	required

Returns:

Type	Description
`float`	The same time in seconds (the package-wide parameter convention).

Source code in src/dapkel/core/timing.py

def to_s(picoseconds: float) -> float:
    """Convert picoseconds to seconds.

    Parameters
    ----------
    picoseconds : float
        A time in picoseconds (the internal working unit).

    Returns
    -------
    float
        The same time in seconds (the package-wide parameter convention).
    """
    return float(picoseconds) * 1e-12

codes_for_full_scale ¶

codes_for_full_scale(full_scale_s: float, *, code_s: float = CODE_S, headroom: float = 1.1) -> int

Size the TDC code axis for a given full scale.

The ring sweeps full_scale_s / code_s codes per measurement range; the headroom covers per-pixel spread in the ring frequency, and the code axis is trimmed to what is actually populated afterwards anyway.

Parameters:

Name	Type	Description	Default
`full_scale_s`	`float`	TDC full scale (one reference-clock period), in seconds.	required
`code_s`	`float`	Nominal width of one code, in seconds. The default is 'CODE_S'.	`CODE_S`
`headroom`	`float`	Multiplier applied to the nominal code count. The default is 1.1.	`1.1`

Returns:

Type	Description
`int`	Number of code bins, capped at `MAX_TS_CODE + 1`.

Raises:

Type	Description
`ValueError`	Raised when `full_scale_s` or `code_s` is not positive.

Warns:

Type	Description
`UserWarning`	Raised when the requested full scale exceeds what the coarse counter can express, i.e. the reference clock is too slow and codes will wrap.

Source code in src/dapkel/core/timing.py

def codes_for_full_scale(
    full_scale_s: float, *, code_s: float = CODE_S, headroom: float = 1.1
) -> int:
    """Size the TDC code axis for a given full scale.

    The ring sweeps ``full_scale_s / code_s`` codes per measurement range; the
    headroom covers per-pixel spread in the ring frequency, and the code axis is
    trimmed to what is actually populated afterwards anyway.

    Parameters
    ----------
    full_scale_s : float
        TDC full scale (one reference-clock period), in seconds.
    code_s : float, optional
        Nominal width of one code, in seconds. The default is 'CODE_S'.
    headroom : float, optional
        Multiplier applied to the nominal code count. The default is 1.1.

    Returns
    -------
    int
        Number of code bins, capped at ``MAX_TS_CODE + 1``.

    Raises
    ------
    ValueError
        Raised when ``full_scale_s`` or ``code_s`` is not positive.

    Warns
    -----
    UserWarning
        Raised when the requested full scale exceeds what the coarse counter
        can express, i.e. the reference clock is too slow and codes will wrap.
    """
    if full_scale_s <= 0:
        raise ValueError(f"full_scale_s must be positive, got {full_scale_s}")
    if code_s <= 0:
        raise ValueError(f"code_s must be positive, got {code_s}")

    nominal = full_scale_s / code_s
    if nominal > MAX_TS_CODE:
        warnings.warn(
            f"A full scale of {full_scale_s * 1e9:.1f} ns needs "
            f"~{nominal:.0f} codes, but the {COARSE_BITS}-bit coarse counter "
            f"saturates at {MAX_TS_CODE} "
            f"(~{MAX_TS_CODE * code_s * 1e9:.0f} ns). The reference clock is "
            f"too slow - below ~{1e-6 / (MAX_TS_CODE * code_s):.2f} MHz the "
            "codes wrap and the timestamps are meaningless.",
            UserWarning,
            stacklevel=2,
        )
    return min(int(nominal * headroom) + 1, MAX_TS_CODE + 1)

resolve_tdc_full_scale_s ¶

resolve_tdc_full_scale_s(*, time_lut=None, full_scale_s: float | None = None, open_shutter_time_s: float | None = None) -> tuple[float, str]

Return the TDC full scale to model the accidental background on.

The timestamp is a code on the TDC's measurement range, so a difference of two timestamps is bounded by that range and - phases being uniform within it - distributed as a triangle of that half-base. Resolution order, most to least trustworthy:

time_lut - a calibration LUT is self-describing: its largest entry is the full scale it was built for.
full_scale_s - stated outright by the caller.
open_shutter_time_s - a fallback that is only correct when the shutter is exactly one clock period. Warns, because a shutter longer than the clock period folds (see the module docstring).

Parameters:

Name	Type	Description	Default
`time_lut`	`ndarray \| None`	A `(n_codes,)` or `(32, 32, n_codes)` code -> ps LUT. The default is None.	`None`
`full_scale_s`	`float \| None`	TDC full scale in seconds, stated explicitly. The default is None.	`None`
`open_shutter_time_s`	`float \| None`	Photon-sensitive window per frame, in seconds, used only as a last resort. The default is None.	`None`

Returns:

Type	Description
`tuple[float, str]`	The full scale in seconds and a human-readable source string.

Raises:

Type	Description
`ValueError`	Raised when none of the three inputs is given.

Warns:

Type	Description
`UserWarning`	Raised when the full scale had to be guessed from the shutter, or when a LUT and an explicit `full_scale_s` disagree by more than 5 %.

Source code in src/dapkel/core/timing.py

def resolve_tdc_full_scale_s(
    *,
    time_lut=None,
    full_scale_s: float | None = None,
    open_shutter_time_s: float | None = None,
) -> tuple[float, str]:
    """Return the TDC full scale to model the accidental background on.

    The timestamp is a code on the TDC's measurement range, so a difference of
    two timestamps is bounded by that range and - phases being uniform within it
    - distributed as a triangle of that half-base. Resolution order, most to
    least trustworthy:

    1. ``time_lut`` - a calibration LUT is self-describing: its largest entry
       *is* the full scale it was built for.
    2. ``full_scale_s`` - stated outright by the caller.
    3. ``open_shutter_time_s`` - a fallback that is only correct when the
       shutter is exactly one clock period. Warns, because a shutter longer than
       the clock period folds (see the module docstring).

    Parameters
    ----------
    time_lut : np.ndarray | None, optional
        A ``(n_codes,)`` or ``(32, 32, n_codes)`` code -> ps LUT. The default
        is None.
    full_scale_s : float | None, optional
        TDC full scale in seconds, stated explicitly. The default is None.
    open_shutter_time_s : float | None, optional
        Photon-sensitive window per frame, in seconds, used only as a last
        resort. The default is None.

    Returns
    -------
    tuple[float, str]
        The full scale in seconds and a human-readable source string.

    Raises
    ------
    ValueError
        Raised when none of the three inputs is given.

    Warns
    -----
    UserWarning
        Raised when the full scale had to be guessed from the shutter, or when
        a LUT and an explicit ``full_scale_s`` disagree by more than 5 %.
    """
    lut_s = None
    if time_lut is not None:
        lut_s = to_s(float(time_lut.max()))

    if lut_s is not None and full_scale_s is not None:
        if abs(lut_s - full_scale_s) > 0.05 * lut_s:
            warnings.warn(
                f"full_scale_s={full_scale_s * 1e9:.1f} ns disagrees with the "
                f"LUT's own full scale of {lut_s * 1e9:.1f} ns. Using the LUT "
                "- it is the table the timestamps were actually converted "
                "with. Pass the matching calibration instead.",
                UserWarning,
                stacklevel=2,
            )
        return lut_s, f"LUT full scale ({lut_s * 1e9:.2f} ns)"

    if lut_s is not None:
        return lut_s, f"LUT full scale ({lut_s * 1e9:.2f} ns)"
    if full_scale_s is not None:
        return float(full_scale_s), f"stated ({full_scale_s * 1e9:.2f} ns)"
    if open_shutter_time_s is not None:
        warnings.warn(
            "No calibration LUT and no full_scale_s, so the TDC full scale is "
            f"taken from open_shutter_time_s="
            f"{open_shutter_time_s * 1e9:.1f} ns. That is only right when the "
            "shutter is exactly one reference-clock period; a longer shutter "
            "folds the timestamps into the clock period and the accidental "
            "triangle is narrower than this. Pass full_scale_s to be sure.",
            UserWarning,
            stacklevel=2,
        )
        return (
            float(open_shutter_time_s),
            f"assumed = open shutter time ({open_shutter_time_s * 1e9:.1f} ns)",
        )

    raise ValueError(
        "Cannot resolve the TDC full scale: pass a time_lut, or full_scale_s, "
        "or open_shutter_time_s (in seconds)."
    )

resolve_cycle_time ¶

resolve_cycle_time(firmware_version: str, exp_time_s: float | None = None) -> tuple[float, str]

Return the length of one acquisition cycle and where it came from.

This is the wall-clock repetition period of a frame, which is what a photon rate divides by: nframes * n_files * cycle is the elapsed time of a run.

Computed from the firmware setting, and never from the frame_rate_cnt.txt the acquisition exe writes beside the data: that exe is not ours, so no analysis here is allowed to depend on its output. The cycle length is a property of the firmware and the requested exposure, and Kelpie_run.m states it directly: exp_time_s = 0 -> 9 us, 1e-6 -> 10 us, 2e-6 -> 11 us.

For the record, the counter is self-consistent where it exists - it is a 200 MHz tick count over the whole run, latched by the firmware, and across every sample folder it fits ticks = nframes * N + 138 + window/5ns exactly. What it says is that the real cycle is 9.685-9.770 µs, i.e. 685-770 ns per frame longer than the nominal 9 µs used here, and that the 50-500 ns exposure window sits inside the cycle rather than extending it. So a rate normalised by the nominal 9 µs runs ~7.8% high. Whether to prefer the nominal or a measured cycle is a decision for the experiment, not for this function to make silently off a file it happens to find on disk.

Parameters:

Name	Type	Description	Default
`firmware_version`	`str`	`'short_exposure'` (the cycle is always 9 µs) or `'long_exposure'` (the cycle is `exp_time_s` + 9 µs). `'short_window'` and `'full_window'` are accepted as aliases of the two.	required
`exp_time_s`	`float \| None`	The exposure X requested of long-exposure firmware, in seconds. The firmware adds 9 µs to it. Required for `'long_exposure'`, ignored for `'short_exposure'`. The default is None.	`None`

Returns:

Type	Description
`tuple[float, str]`	The cycle length in seconds and a human-readable source string.

Raises:

Type	Description
`ValueError`	Raised on an unknown `firmware_version`, or when `'long_exposure'` is given without `exp_time_s`.

Source code in src/dapkel/core/timing.py

def resolve_cycle_time(
    firmware_version: str, exp_time_s: float | None = None
) -> tuple[float, str]:
    """Return the length of one acquisition cycle and where it came from.

    This is the wall-clock repetition period of a frame, which is what a
    photon *rate* divides by: ``nframes * n_files * cycle`` is the elapsed
    time of a run.

    Computed from the firmware setting, and **never** from the
    ``frame_rate_cnt.txt`` the acquisition exe writes beside the data: that exe
    is not ours, so no analysis here is allowed to depend on its output. The
    cycle length is a property of the firmware and the requested exposure, and
    ``Kelpie_run.m`` states it directly: ``exp_time_s = 0 -> 9 us``,
    ``1e-6 -> 10 us``, ``2e-6 -> 11 us``.

    For the record, the counter *is* self-consistent where it exists - it is a
    200 MHz tick count over the whole run, latched by the firmware, and across
    every sample folder it fits ``ticks = nframes * N + 138 + window/5ns``
    exactly. What it says is that the real cycle is **9.685-9.770 µs**, i.e.
    685-770 ns per frame longer than the nominal 9 µs used here, and that the
    50-500 ns exposure window sits *inside* the cycle rather than extending it.
    So a rate normalised by the nominal 9 µs runs ~7.8% high. Whether to prefer
    the nominal or a measured cycle is a decision for the experiment, not for
    this function to make silently off a file it happens to find on disk.

    Parameters
    ----------
    firmware_version : str
        ``'short_exposure'`` (the cycle is always 9 µs) or ``'long_exposure'``
        (the cycle is ``exp_time_s`` + 9 µs). ``'short_window'`` and
        ``'full_window'`` are accepted as aliases of the two.
    exp_time_s : float | None, optional
        The exposure X requested of long-exposure firmware, in seconds. The
        firmware adds 9 µs to it. Required for ``'long_exposure'``, ignored for
        ``'short_exposure'``. The default is None.

    Returns
    -------
    tuple[float, str]
        The cycle length in seconds and a human-readable source string.

    Raises
    ------
    ValueError
        Raised on an unknown ``firmware_version``, or when
        ``'long_exposure'`` is given without ``exp_time_s``.
    """
    try:
        version = _CYCLE_ALIASES[firmware_version]
    except KeyError:
        raise ValueError(
            "firmware_version must be 'short_exposure' or 'long_exposure' "
            f"(aliases 'short_window' / 'full_window'), got "
            f"{firmware_version!r}"
        ) from None

    base = FREE_RUNNING_US * 1e-6
    if version == "short_exposure":
        return base, f"short_exposure: {FREE_RUNNING_US:.0f} µs cycle"
    if exp_time_s is None:
        raise ValueError(
            "firmware_version='long_exposure' needs exp_time_s (the exposure X "
            "requested, in seconds; the firmware adds "
            f"{FREE_RUNNING_US:.0f} µs to it)."
        )
    return (
        exp_time_s + base,
        f"long_exposure: {exp_time_s * 1e6:.3f} µs + "
        f"{FREE_RUNNING_US:.0f} µs = {(exp_time_s + base) * 1e6:.3f} µs cycle",
    )

resolve_frame_time ¶

resolve_frame_time(explicit_s: float | None, nframes: int | None = None) -> tuple[float, str]

Return the per-frame period and a description of where it came from.

explicit exp_time_s supplied -> exp_time_s + 9 µs overhead
fallback -> 9 µs free-running base

The acquisition exe also writes a frame_rate_cnt.txt beside the data - a 200 MHz tick count for the whole run, latched by the firmware. It is deliberately not read here. It is written by an exe we do not control, so nothing in the analysis path is allowed to depend on it; the period comes from the settings the run was started with. See 'resolve_cycle_time' for what the counter does contain, and for the measured-vs-nominal gap.

Parameters:

Name	Type	Description	Default
`explicit_s`	`float \| None`	Explicit exposure time in seconds, or None for the free-running base.	required
`nframes`	`int \| None`	Unused. Kept so existing callers do not break; it was only ever needed to turn the counter's tick count into a period. The default is None.	`None`

Returns:

Type	Description
`tuple[float, str]`	The frame period in seconds and a human-readable source string.

Source code in src/dapkel/core/timing.py

def resolve_frame_time(
    explicit_s: float | None, nframes: int | None = None
) -> tuple[float, str]:
    """Return the per-frame period and a description of where it came from.

    1. explicit ``exp_time_s`` supplied -> ``exp_time_s`` + 9 µs overhead
    2. fallback                         -> 9 µs free-running base

    The acquisition exe also writes a ``frame_rate_cnt.txt`` beside the data -
    a 200 MHz tick count for the whole run, latched by the firmware. **It is
    deliberately not read here.** It is written by an exe we do not control, so
    nothing in the analysis path is allowed to depend on it; the period comes
    from the settings the run was started with. See 'resolve_cycle_time' for
    what the counter does contain, and for the measured-vs-nominal gap.

    Parameters
    ----------
    explicit_s : float | None
        Explicit exposure time in seconds, or None for the free-running base.
    nframes : int | None, optional
        Unused. Kept so existing callers do not break; it was only ever needed
        to turn the counter's tick count into a period. The default is None.

    Returns
    -------
    tuple[float, str]
        The frame period in seconds and a human-readable source string.
    """
    if explicit_s is not None:
        ft = explicit_s + FREE_RUNNING_US * 1e-6
        return (
            ft,
            f"exp_time_s={explicit_s * 1e6:.1f} µs + "
            f"{FREE_RUNNING_US:.0f} µs overhead",
        )

    ft = FREE_RUNNING_US * 1e-6
    return ft, f"free-running base ({FREE_RUNNING_US} µs)"

resolve_live_time ¶

resolve_live_time(firmware_version: str, nframes: int, *, open_shutter_time_s: float | None = None, exp_time_s: float | None = None) -> tuple[float | None, str]

Return the live seconds per frame for the rate normalisation.

The photon rate is counts / (nframes * n_files * live_per_frame). How much of each frame is photon-sensitive depends on the firmware:

'short_window' - the current firmware: each ~9 µs frame is mostly readout and only the open shutter window (open_shutter_time_s, e.g. 200e-9 s, < 9 µs) is live. Returns (None, ...) when open_shutter_time_s is not supplied, so the caller can decide whether that is an error (DCR) or just "no rate available" (hitmap).
'full_window' - the legacy firmware: the whole frame is live, so the per-frame time comes from 'resolve_frame_time'.

Parameters:

Name	Type	Description	Default
`firmware_version`	`str`	`'short_window'` or `'full_window'`.	required
`nframes`	`int`	Frames per file (used by the `'full_window'` resolution).	required
`open_shutter_time_s`	`float \| None`	Photon-sensitive window per frame, in seconds (`'short_window'` only). The default is None.	`None`
`exp_time_s`	`float \| None`	Legacy exposure time, in seconds (`'full_window'` only). The default is None.	`None`

Returns:

Type	Description
`tuple[float \| None, str]`	The live seconds per frame (None if unresolved) and a human-readable source string.

Raises:

Type	Description
`ValueError`	Raised when `firmware_version` is not one of the two known values.

Source code in src/dapkel/core/timing.py

def resolve_live_time(
    firmware_version: str,
    nframes: int,
    *,
    open_shutter_time_s: float | None = None,
    exp_time_s: float | None = None,
) -> tuple[float | None, str]:
    """Return the live seconds per frame for the rate normalisation.

    The photon rate is ``counts / (nframes * n_files * live_per_frame)``. How
    much of each frame is photon-sensitive depends on the firmware:

    * ``'short_window'`` - the current firmware: each ~9 µs frame is mostly
      readout and only the open shutter window (``open_shutter_time_s``, e.g.
      ``200e-9`` s, ``< 9 µs``) is live. Returns ``(None, ...)`` when
      ``open_shutter_time_s`` is not supplied, so the caller can decide whether
      that is an error (DCR) or just "no rate available" (hitmap).

    * ``'full_window'`` - the legacy firmware: the whole frame is live, so the
      per-frame time comes from 'resolve_frame_time'.

    Parameters
    ----------
    firmware_version : str
        ``'short_window'`` or ``'full_window'``.
    nframes : int
        Frames per file (used by the ``'full_window'`` resolution).
    open_shutter_time_s : float | None, optional
        Photon-sensitive window per frame, in seconds (``'short_window'``
        only). The default is None.
    exp_time_s : float | None, optional
        Legacy exposure time, in seconds (``'full_window'`` only). The default
        is None.

    Returns
    -------
    tuple[float | None, str]
        The live seconds per frame (None if unresolved) and a human-readable
        source string.

    Raises
    ------
    ValueError
        Raised when ``firmware_version`` is not one of the two known values.
    """
    if firmware_version == "short_window":
        if open_shutter_time_s is None:
            return None, "short_window (no open_shutter_time_s given)"
        return (
            open_shutter_time_s,
            f"short_window: {open_shutter_time_s * 1e9:.1f} ns window",
        )
    if firmware_version == "full_window":
        return resolve_frame_time(exp_time_s, nframes)
    raise ValueError(
        "firmware_version must be 'short_window' or 'full_window', got "
        f"{firmware_version!r}"
    )

`core.pairs` — which pixel pairs, and what they are called¶

pairs ¶

Which pixel pairs an analysis is about, and what they are called.

Every coincidence analysis starts by turning two detector groups into an ordered list of pairs and a label per pair. That enumeration decides the row order of every stage-1 artifact, so two analyses are only comparable if they build it the same way - which is why it lives here rather than in each of them. 'delta_t' had it twice: once inline in the feather path, once as a private helper for the counts path.

The label format "ra,ca-rb,cb" is a stored contract: it is the key 'calc_diff.calculate_differences' returns, the column name in the delta-t feather, and the labels entry in every counts sidecar. Do not reformat it.

pair_label ¶

pair_label(a: Pixel, b: Pixel) -> str

Return the canonical "ra,ca-rb,cb" name of one pixel pair.

Source code in src/dapkel/core/pairs.py

def pair_label(a: Pixel, b: Pixel) -> str:
    """Return the canonical ``"ra,ca-rb,cb"`` name of one pixel pair."""
    return f"{a[0]},{a[1]}-{b[0]},{b[1]}"

pair_list ¶

pair_list(pixels: Sequence[Sequence[Pixel]], mode: str = 'all_pairs') -> list[tuple[Pixel, Pixel]]

Enumerate the ordered (pixel_a, pixel_b) pairs of two groups.

Parameters:

Name	Type	Description	Default
`pixels`	`Sequence[Sequence[tuple[int, int]]]`	Two groups `[group_a, group_b]`; a bare `(row, col)` is accepted for a one-pixel group.	required
`mode`	`str`	`'all_pairs'` (default) for every A-B combination, or `'1v1'` to zip the groups element-wise.	`'all_pairs'`

Returns:

Type	Description
`list[tuple[tuple[int, int], tuple[int, int]]]`	The pairs, in the order every artifact stores them in.

Raises:

Type	Description
`ValueError`	Raised on an unknown `mode`, or unequal groups under `'1v1'`.

Source code in src/dapkel/core/pairs.py

def pair_list(
    pixels: Sequence[Sequence[Pixel]], mode: str = "all_pairs"
) -> list[tuple[Pixel, Pixel]]:
    """Enumerate the ordered ``(pixel_a, pixel_b)`` pairs of two groups.

    Parameters
    ----------
    pixels : Sequence[Sequence[tuple[int, int]]]
        Two groups ``[group_a, group_b]``; a bare ``(row, col)`` is accepted
        for a one-pixel group.
    mode : str, optional
        ``'all_pairs'`` (default) for every A-B combination, or ``'1v1'`` to
        zip the groups element-wise.

    Returns
    -------
    list[tuple[tuple[int, int], tuple[int, int]]]
        The pairs, in the order every artifact stores them in.

    Raises
    ------
    ValueError
        Raised on an unknown ``mode``, or unequal groups under ``'1v1'``.
    """
    if mode not in ("all_pairs", "1v1"):
        raise ValueError(f"mode must be 'all_pairs' or '1v1', got {mode!r}")

    group_a = as_pixel_list(pixels[0])
    group_b = as_pixel_list(pixels[1])
    if mode == "1v1":
        if len(group_a) != len(group_b):
            raise ValueError(
                "mode='1v1' needs equal-length groups "
                f"(got {len(group_a)} and {len(group_b)})."
            )
        return list(zip(group_a, group_b, strict=True))
    return [(a, b) for a in group_a for b in group_b]

pair_labels ¶

pair_labels(pixels: Sequence[Sequence[Pixel]], mode: str = 'all_pairs') -> tuple[list[str], list[Pixel]]

Pair labels and the de-duplicated pixels they mention.

The two things a stage-1 loop needs: what to call each output row, and which pixels to decode out of the frames.

Parameters:

Name	Type	Description	Default
`pixels`	`Sequence[Sequence[tuple[int, int]]]`	Two groups `[group_a, group_b]`.	required
`mode`	`str`	`'all_pairs'` (default) or `'1v1'`.	`'all_pairs'`

Returns:

Name	Type	Description
`labels`	`list[str]`	`"ra,ca-rb,cb"` per pair, in 'pair_list' order.
`all_pixels`	`list[tuple[int, int]]`	Every pixel mentioned, de-duplicated, first-seen order.

Raises:

Type	Description
`ValueError`	Raised on an unknown `mode`, or unequal groups under `'1v1'`.

Source code in src/dapkel/core/pairs.py

def pair_labels(
    pixels: Sequence[Sequence[Pixel]], mode: str = "all_pairs"
) -> tuple[list[str], list[Pixel]]:
    """Pair labels and the de-duplicated pixels they mention.

    The two things a stage-1 loop needs: what to call each output row, and
    which pixels to decode out of the frames.

    Parameters
    ----------
    pixels : Sequence[Sequence[tuple[int, int]]]
        Two groups ``[group_a, group_b]``.
    mode : str, optional
        ``'all_pairs'`` (default) or ``'1v1'``.

    Returns
    -------
    labels : list[str]
        ``"ra,ca-rb,cb"`` per pair, in 'pair_list' order.
    all_pixels : list[tuple[int, int]]
        Every pixel mentioned, de-duplicated, first-seen order.

    Raises
    ------
    ValueError
        Raised on an unknown ``mode``, or unequal groups under ``'1v1'``.
    """
    pl = pair_list(pixels, mode)
    labels = [pair_label(a, b) for a, b in pl]
    all_pixels = list(
        dict.fromkeys(as_pixel_list(pixels[0]) + as_pixel_list(pixels[1]))
    )
    return labels, all_pixels

`core.reduce` — folding frames into a map¶

reduce ¶

Stage 1: folding a set of '.bin' files down to one sensor map.

Every count-like analysis in dapkel is the same loop - unpack each file, pull one (32, 32) slab out of the frames, add it up - differing only in which slab it pulls. That loop lives here once; the analyses supply the extraction.

accumulate_frames ¶

accumulate_frames(files: list[str], extract: Extract, *, nframes: int, need_time_series: bool = True, label: str = '', shape: tuple[int, int] = (32, 32)) -> ndarray

Unpack every file and sum an extracted slab over all of them.

Parameters:

Name	Type	Description	Default
`files`	`list[str]`	Paths to the '.bin' files to unpack and accumulate.	required
`extract`	`callable`	`extract(time_series, photon_counts) -> np.ndarray` returning the (32, 32) contribution of one file. Both inputs are `(32, 32, nframes)`; `time_series` is None when `need_time_series` is False. Typical bodies are `lambda ts, pc: pc.sum(axis=2)` for photon counts and `lambda ts, pc: (ts > 0).sum(axis=2)` for ORT occupancy.	required
`nframes`	`int`	Frames per file - the acquisition setting, required rather than inferred. 'io.frames_in_file' is there when a file's size is the only record left, but which frame count a run used is the caller's to state, not this loop's to guess.	required
`need_time_series`	`bool`	Whether `extract` uses the timestamps. Passing False lets 'unpack' skip the coarse/fine time decoding - roughly two thirds of the per-pixel work - so count-only analyses should set it. The default is True.	`True`
`label`	`str`	Label used in the progress bar. The default is "".	`''`
`shape`	`tuple[int, int]`	Shape of the accumulator. The default is `(32, 32)`.	`(32, 32)`

Returns:

Type	Description
`ndarray`	The summed map, float64.

Source code in src/dapkel/core/reduce.py

def accumulate_frames(
    files: list[str],
    extract: Extract,
    *,
    nframes: int,
    need_time_series: bool = True,
    label: str = "",
    shape: tuple[int, int] = (32, 32),
) -> np.ndarray:
    """Unpack every file and sum an extracted slab over all of them.

    Parameters
    ----------
    files : list[str]
        Paths to the '.bin' files to unpack and accumulate.
    extract : callable
        ``extract(time_series, photon_counts) -> np.ndarray`` returning the
        (32, 32) contribution of one file. Both inputs are
        ``(32, 32, nframes)``; ``time_series`` is None when
        ``need_time_series`` is False. Typical bodies are
        ``lambda ts, pc: pc.sum(axis=2)`` for photon counts and
        ``lambda ts, pc: (ts > 0).sum(axis=2)`` for ORT occupancy.
    nframes : int
        Frames per file - the acquisition setting, required rather than
        inferred. 'io.frames_in_file' is there when a file's size is the only
        record left, but which frame count a run used is the caller's to
        state, not this loop's to guess.
    need_time_series : bool, optional
        Whether ``extract`` uses the timestamps. Passing False lets 'unpack'
        skip the coarse/fine time decoding - roughly two thirds of the
        per-pixel work - so count-only analyses should set it. The default is
        True.
    label : str, optional
        Label used in the progress bar. The default is "".
    shape : tuple[int, int], optional
        Shape of the accumulator. The default is ``(32, 32)``.

    Returns
    -------
    np.ndarray
        The summed map, float64.
    """
    total = np.zeros(shape, dtype=np.float64)
    for fp in tqdm(files, desc=label or None):
        ts, pc = unpack(fp, nframes, compute_time_series=need_time_series)
        total += extract(ts, pc)
    return total

assemble_64 ¶

assemble_64(quadrants: dict[str, ndarray], layout: dict[str, tuple[int, int]]) -> ndarray

Interleave four (32, 32) quadrant maps into the (64, 64) sensor map.

Each SPAD tag reads one micropixel per 2x2 macropixel, so the four quadrant maps interleave rather than tile: quadrant with offset (drow, dcol) occupies full[drow::2, dcol::2].

Parameters:

Name	Type	Description	Default
`quadrants`	`dict[str, ndarray]`	Mapping of SPAD tag to its (32, 32) map. Tags absent from the mapping are left as zero, so a partial acquisition still assembles.	required
`layout`	`dict[str, tuple[int, int]]`	Mapping of SPAD tag to its `(drow, dcol)` offset inside the 2x2 macropixel. See the `_SPAD_LAYOUT` of the calling module - the SPAD indices run clockwise (S0 S1 / S3 S2), not row-major.	required

Returns:

Type	Description
`ndarray`	The (64, 64) assembled map, float64.

Source code in src/dapkel/core/reduce.py

def assemble_64(
    quadrants: dict[str, np.ndarray], layout: dict[str, tuple[int, int]]
) -> np.ndarray:
    """Interleave four (32, 32) quadrant maps into the (64, 64) sensor map.

    Each SPAD tag reads one micropixel per 2x2 macropixel, so the four
    quadrant maps interleave rather than tile: quadrant with offset
    ``(drow, dcol)`` occupies ``full[drow::2, dcol::2]``.

    Parameters
    ----------
    quadrants : dict[str, np.ndarray]
        Mapping of SPAD tag to its (32, 32) map. Tags absent from the mapping
        are left as zero, so a partial acquisition still assembles.
    layout : dict[str, tuple[int, int]]
        Mapping of SPAD tag to its ``(drow, dcol)`` offset inside the 2x2
        macropixel. See the ``_SPAD_LAYOUT`` of the calling module - the SPAD
        indices run clockwise (S0 S1 / S3 S2), not row-major.

    Returns
    -------
    np.ndarray
        The (64, 64) assembled map, float64.
    """
    full = np.zeros((64, 64), dtype=np.float64)
    for tag, (drow, dcol) in layout.items():
        if tag in quadrants:
            full[drow::2, dcol::2] = quadrants[tag]
    return full

`core.plots` — the two figure shapes¶

plots ¶

Stage 2: the two figure shapes every analysis draws.

A sensor map (a 2D image of the array) and a sorted-percentile distribution cover every plot in the package bar the TDC QA panel. Both used to be copied per analysis, which is how three heatmaps ended up with three different figure sizes and two different NaN policies.

Figure sizing follows the package rule: square sensor maps set 16x16 in code, everything else takes figure.figsize from the active style (the dapkel style ships 16x10). So 'sensor_map' is explicit and 'sorted_distribution' is not.

All statistics here are NaN-aware. Plain np.median/np.percentile propagate NaN, so a single masked pixel would turn vmax into NaN and flatten the whole image to one colour with no error - see 'docs/adding_an_analysis.md'.

map_stats ¶

map_stats(data: ndarray, *, scale: float = 1.0) -> tuple[float, float, float]

Return the NaN-aware (median, max, total) of a sensor map.

Parameters:

Name	Type	Description	Default
`data`	`ndarray`	The map.	required
`scale`	`float`	Multiplied into the data first, for unit conversion (e.g. 100.0 to turn a ratio into a percentage). The default is 1.0.	`1.0`

Returns:

Type	Description
`tuple[float, float, float]`	Median, maximum and sum, all ignoring NaN. An all-NaN map gives `(nan, nan, 0.0)` rather than raising.

Source code in src/dapkel/core/plots.py

def map_stats(
    data: np.ndarray, *, scale: float = 1.0
) -> tuple[float, float, float]:
    """Return the NaN-aware ``(median, max, total)`` of a sensor map.

    Parameters
    ----------
    data : np.ndarray
        The map.
    scale : float, optional
        Multiplied into the data first, for unit conversion (e.g. 100.0 to
        turn a ratio into a percentage). The default is 1.0.

    Returns
    -------
    tuple[float, float, float]
        Median, maximum and sum, all ignoring NaN. An all-NaN map gives
        ``(nan, nan, 0.0)`` rather than raising.
    """
    scaled = data * scale
    if not np.any(np.isfinite(scaled)):
        return float("nan"), float("nan"), 0.0
    return (
        float(np.nanmedian(scaled)),
        float(np.nanmax(scaled)),
        float(np.nansum(scaled)),
    )

sensor_map ¶

sensor_map(data: ndarray, *, title: str, clabel: str | None = None, cmap: str | None = None, scale: float = 1.0, xlabel: str = 'Column', ylabel: str = 'Row') -> Figure

Render a sensor map as a square 2D image with a colour bar.

The colour scale is clipped at the 99th percentile so a handful of hot pixels cannot wash out the rest of the array.

Parameters:

Name	Type	Description	Default
`data`	`ndarray`	The (rows, cols) map to draw.	required
`title`	`str`	Full figure title. Callers build this themselves - the wording is measurement-specific - typically from 'map_stats'.	required
`clabel`	`str \| None`	Colour-bar label. The default is None (no label).	`None`
`cmap`	`str \| None`	Matplotlib colormap name. The default is None - use the active style's `image.cmap`.	`None`
`scale`	`float`	Multiplied into the data before drawing. The default is 1.0.	`1.0`
`xlabel`	`str`	Axis labels. The defaults are `'Column'` and `'Row'`.	`'Column'`
`ylabel`	`str`	Axis labels. The defaults are `'Column'` and `'Row'`.	`'Column'`

Returns:

Type	Description
`Figure`	The figure. Nothing is written to disk.

Source code in src/dapkel/core/plots.py

def sensor_map(
    data: np.ndarray,
    *,
    title: str,
    clabel: str | None = None,
    cmap: str | None = None,
    scale: float = 1.0,
    xlabel: str = "Column",
    ylabel: str = "Row",
) -> plt.Figure:
    """Render a sensor map as a square 2D image with a colour bar.

    The colour scale is clipped at the 99th percentile so a handful of hot
    pixels cannot wash out the rest of the array.

    Parameters
    ----------
    data : np.ndarray
        The (rows, cols) map to draw.
    title : str
        Full figure title. Callers build this themselves - the wording is
        measurement-specific - typically from 'map_stats'.
    clabel : str | None, optional
        Colour-bar label. The default is None (no label).
    cmap : str | None, optional
        Matplotlib colormap name. The default is None - use the active
        style's ``image.cmap``.
    scale : float, optional
        Multiplied into the data before drawing. The default is 1.0.
    xlabel, ylabel : str, optional
        Axis labels. The defaults are ``'Column'`` and ``'Row'``.

    Returns
    -------
    plt.Figure
        The figure. Nothing is written to disk.
    """
    scaled = data * scale
    vmax = (
        float(np.nanpercentile(scaled, _VMAX_PERCENTILE))
        if np.any(np.isfinite(scaled))
        else None
    )

    # Square figure for the sensor map (kept explicit, per the package rule);
    # fonts and colormap come from the active matplotlib style.
    fig, ax = plt.subplots(figsize=SENSOR_MAP_FIGSIZE)

    im = ax.imshow(
        scaled, cmap=cmap, aspect="equal", vmax=vmax, origin="lower"
    )
    fig.colorbar(im, ax=ax, fraction=0.046, pad=0.04, label=clabel)

    ax.set_title(title)
    ax.set_xlabel(xlabel)
    ax.set_ylabel(ylabel)
    return fig

sorted_distribution ¶

sorted_distribution(data: ndarray, *, ylabel: str, title: str, xlabel: str = 'Pixel percentile  (%)', logy: bool = False, scale: float = 1.0, fmt: str = '.3g', unit: str = '') -> Figure

Plot per-pixel values sorted low-to-high against their percentile.

Non-finite values are dropped before sorting, so masked or undefined pixels neither plot nor skew the median and mean lines.

Parameters:

Name	Type	Description	Default
`data`	`ndarray`	The map whose values are ranked.	required
`ylabel`	`str`	Y-axis label, including the unit (e.g. `'DCR (cps)'`).	required
`title`	`str`	Figure title.	required
`xlabel`	`str`	X-axis label. The default is `'Pixel percentile (%)'`.	`'Pixel percentile (%)'`
`logy`	`bool`	Use a logarithmic y-axis, appropriate when the values span decades (DCR does). The default is False.	`False`
`scale`	`float`	Multiplied into the data first. The default is 1.0.	`1.0`
`fmt`	`str`	Format spec for the median/mean values in the legend. The default is `'.3g'`.	`'.3g'`
`unit`	`str`	Unit appended to the legend entries (e.g. `'cps'`, `'%'`). The default is "".	`''`

Returns:

Type	Description
`Figure`	The figure. Nothing is written to disk.

Source code in src/dapkel/core/plots.py

def sorted_distribution(
    data: np.ndarray,
    *,
    ylabel: str,
    title: str,
    xlabel: str = "Pixel percentile  (%)",
    logy: bool = False,
    scale: float = 1.0,
    fmt: str = ".3g",
    unit: str = "",
) -> plt.Figure:
    """Plot per-pixel values sorted low-to-high against their percentile.

    Non-finite values are dropped before sorting, so masked or undefined
    pixels neither plot nor skew the median and mean lines.

    Parameters
    ----------
    data : np.ndarray
        The map whose values are ranked.
    ylabel : str
        Y-axis label, including the unit (e.g. ``'DCR  (cps)'``).
    title : str
        Figure title.
    xlabel : str, optional
        X-axis label. The default is ``'Pixel percentile  (%)'``.
    logy : bool, optional
        Use a logarithmic y-axis, appropriate when the values span decades
        (DCR does). The default is False.
    scale : float, optional
        Multiplied into the data first. The default is 1.0.
    fmt : str, optional
        Format spec for the median/mean values in the legend. The default is
        ``'.3g'``.
    unit : str, optional
        Unit appended to the legend entries (e.g. ``'cps'``, ``'%'``). The
        default is "".

    Returns
    -------
    plt.Figure
        The figure. Nothing is written to disk.
    """
    values = (data * scale).ravel()
    values = values[np.isfinite(values)]

    # Figure size comes from the active style (non-square), per the package
    # rule; only the square sensor maps set their own.
    fig, ax = plt.subplots()

    suffix = f" {unit}" if unit else ""
    if values.size:
        median = float(np.median(values))
        mean = float(values.mean())
        ordered = np.sort(values)
        percentile = np.linspace(0, 100, ordered.size)

        plot = ax.semilogy if logy else ax.plot
        plot(
            percentile, ordered, ".", markersize=2, alpha=0.7,
            label="_nolegend_",
        )
        ax.axhline(
            median, linestyle="--", linewidth=2,
            label=f"Median  {median:{fmt}}{suffix}",
        )
        ax.axhline(
            mean, linestyle="--", linewidth=1,
            label=f"Mean    {mean:{fmt}}{suffix}",
        )
        ax.legend()

    ax.set_xlim(0, 100)
    ax.set_xticks([0, 25, 50, 75, 100])
    ax.set_xlabel(xlabel)
    ax.set_ylabel(ylabel)
    ax.set_title(title)
    ax.grid(True, which="both", linewidth=0.5, alpha=0.6)
    fig.tight_layout()
    return fig

`core.store` — the stage 1 / stage 2 boundary¶

store ¶

Persistence: the boundary between stage 1 (data) and stage 2 (figures).

The layout under a data folder is::

<data folder>/
  *.bin            stage 0 - raw, read-only, dapkel never writes here
  processed/       stage 1 - arrays ('.npy') + metadata ('.meta.json')
  results/<kind>/  stage 2 - figures only

One rule follows from it: stage 2 gets its data by loading a stage-1 artifact, never by re-unpacking '.bin' files. Retrying a colormap costs a 'load_map', not a full re-read of the raw data.

Every array is saved with a .meta.json sidecar recording how it was acquired (frame counts, live time, tag), so a saved map can still be turned into a rate months later without the original call arguments.

See 'docs/adding_an_analysis.md' for the contract.

processed_dir ¶

processed_dir(folder: str, *, create: bool = True) -> str

Return (and by default create) the processed/ folder of a dataset.

Parameters:

Name	Type	Description	Default
`folder`	`str`	The data folder holding the '.bin' files.	required
`create`	`bool`	Create the folder if missing. The default is True.	`True`

Returns:

Type	Description
`str`	Path to `<folder>/processed`.

Source code in src/dapkel/core/store.py

def processed_dir(folder: str, *, create: bool = True) -> str:
    """Return (and by default create) the ``processed/`` folder of a dataset.

    Parameters
    ----------
    folder : str
        The data folder holding the '.bin' files.
    create : bool, optional
        Create the folder if missing. The default is True.

    Returns
    -------
    str
        Path to ``<folder>/processed``.
    """
    out = os.path.join(folder, PROCESSED_DIR)
    if create:
        os.makedirs(out, exist_ok=True)
    return out

results_dir ¶

results_dir(folder: str, kind: str, *, create: bool = True) -> str

Return (and by default create) the figure folder for one analysis.

Parameters:

Name	Type	Description	Default
`folder`	`str`	The data folder holding the '.bin' files.	required
`kind`	`str`	Analysis name, used as the sub-folder (e.g. `'dcr'`, `'hitmap'`).	required
`create`	`bool`	Create the folder if missing. The default is True.	`True`

Returns:

Type	Description
`str`	Path to `<folder>/results/<kind>`.

Source code in src/dapkel/core/store.py

def results_dir(folder: str, kind: str, *, create: bool = True) -> str:
    """Return (and by default create) the figure folder for one analysis.

    Parameters
    ----------
    folder : str
        The data folder holding the '.bin' files.
    kind : str
        Analysis name, used as the sub-folder (e.g. ``'dcr'``, ``'hitmap'``).
    create : bool, optional
        Create the folder if missing. The default is True.

    Returns
    -------
    str
        Path to ``<folder>/results/<kind>``.
    """
    out = os.path.join(folder, RESULTS_DIR, kind)
    if create:
        os.makedirs(out, exist_ok=True)
    return out

map_file_name ¶

map_file_name(dataset: str, kind: str, tag: str = '') -> str

Build the '.npy' file name for a stage-1 array.

Parameters:

Name	Type	Description	Default
`dataset`	`str`	Name of the dataset, normally the data folder's base name.	required
`kind`	`str`	Analysis name (e.g. `'dcr'`, `'hitmap'`, `'crosstalk'`).	required
`tag`	`str`	Readout/quadrant tag distinguishing artifacts of the same kind. The default is "".	`''`

Returns:

Type	Description
`str`	`'<dataset>_<tag>_<kind>.npy'`, or `'<dataset>_<kind>.npy'` when no tag is given.

Source code in src/dapkel/core/store.py

def map_file_name(dataset: str, kind: str, tag: str = "") -> str:
    """Build the '.npy' file name for a stage-1 array.

    Parameters
    ----------
    dataset : str
        Name of the dataset, normally the data folder's base name.
    kind : str
        Analysis name (e.g. ``'dcr'``, ``'hitmap'``, ``'crosstalk'``).
    tag : str, optional
        Readout/quadrant tag distinguishing artifacts of the same kind. The
        default is "".

    Returns
    -------
    str
        ``'<dataset>_<tag>_<kind>.npy'``, or ``'<dataset>_<kind>.npy'`` when
        no tag is given.
    """
    parts = [dataset, tag.lower(), kind] if tag else [dataset, kind]
    return "_".join(parts) + ".npy"

meta_path ¶

meta_path(npy_path: str) -> str

Return the sidecar '.meta.json' path for a saved '.npy'.

Source code in src/dapkel/core/store.py

def meta_path(npy_path: str) -> str:
    """Return the sidecar '.meta.json' path for a saved '.npy'."""
    return npy_path.removesuffix(".npy") + ".meta.json"

read_meta ¶

read_meta(npy_path: str) -> dict

Read a saved array's '.meta.json' sidecar.

Parameters:

Name	Type	Description	Default
`npy_path`	`str`	Path to the saved '.npy'.	required

Returns:

Type	Description
`dict`	The metadata, or an empty dict when the sidecar is missing or unreadable - callers treat metadata as best-effort.

Source code in src/dapkel/core/store.py

def read_meta(npy_path: str) -> dict:
    """Read a saved array's '.meta.json' sidecar.

    Parameters
    ----------
    npy_path : str
        Path to the saved '.npy'.

    Returns
    -------
    dict
        The metadata, or an empty dict when the sidecar is missing or
        unreadable - callers treat metadata as best-effort.
    """
    path = meta_path(npy_path)
    if os.path.isfile(path):
        try:
            with open(path) as fh:
                return json.load(fh)
        except (OSError, ValueError):
            pass
    return {}

save_map ¶

save_map(data: ndarray, folder: str, *, kind: str, tag: str = '', meta: dict | None = None, file_name: str | None = None, quiet: bool = False) -> str

Save a stage-1 array into processed/ with a metadata sidecar.

Parameters:

Name	Type	Description	Default
`data`	`ndarray`	The array to save.	required
`folder`	`str`	The data folder; the array lands in its `processed/` sub-folder.	required
`kind`	`str`	Analysis name (e.g. `'dcr'`), used in the file name.	required
`tag`	`str`	Readout/quadrant tag. The default is "".	`''`
`meta`	`dict \| None`	Acquisition metadata written alongside as '.meta.json'. The default is None (an empty sidecar is still written, recording kind and tag).	`None`
`file_name`	`str \| None`	Explicit '.npy' file name, overriding the derived one. Used where a naming scheme is already established (the TDC LUTs). The default is None.	`None`
`quiet`	`bool`	Suppress the "saved to" printout. The default is False.	`False`

Returns:

Type	Description
`str`	Path the array was saved to.

Source code in src/dapkel/core/store.py

def save_map(
    data: np.ndarray,
    folder: str,
    *,
    kind: str,
    tag: str = "",
    meta: dict | None = None,
    file_name: str | None = None,
    quiet: bool = False,
) -> str:
    """Save a stage-1 array into ``processed/`` with a metadata sidecar.

    Parameters
    ----------
    data : np.ndarray
        The array to save.
    folder : str
        The data folder; the array lands in its ``processed/`` sub-folder.
    kind : str
        Analysis name (e.g. ``'dcr'``), used in the file name.
    tag : str, optional
        Readout/quadrant tag. The default is "".
    meta : dict | None, optional
        Acquisition metadata written alongside as '.meta.json'. The default
        is None (an empty sidecar is still written, recording kind and tag).
    file_name : str | None, optional
        Explicit '.npy' file name, overriding the derived one. Used where a
        naming scheme is already established (the TDC LUTs). The default is
        None.
    quiet : bool, optional
        Suppress the "saved to" printout. The default is False.

    Returns
    -------
    str
        Path the array was saved to.
    """
    out_dir = processed_dir(folder)
    dataset = os.path.basename(os.path.normpath(folder))
    name = file_name or map_file_name(dataset, kind, tag)
    out_path = os.path.join(out_dir, name)

    np.save(out_path, data)
    with open(meta_path(out_path), "w") as fh:
        json.dump({"kind": kind, "tag": tag, **(meta or {})}, fh, indent=2)

    if not quiet:
        print(f"\n> > > Saved to {out_path} < < <")
    return out_path

load_map ¶

load_map(folder: str | None = None, *, kind: str = '', tag: str = '', npy_path: str | None = None, file_name: str | None = None) -> tuple[ndarray, dict]

Load a stage-1 array and its metadata back from processed/.

Parameters:

Name	Type	Description	Default
`folder`	`str \| None`	The data folder to look under. May be None when `npy_path` is given. The default is None.	`None`
`kind`	`str`	Analysis name, used to derive the file name. The default is "".	`''`
`tag`	`str`	Readout/quadrant tag. The default is "".	`''`
`npy_path`	`str \| None`	Explicit path to a saved '.npy', bypassing the derivation entirely. The default is None.	`None`
`file_name`	`str \| None`	Explicit file name within `processed/`. The default is None.	`None`

Returns:

Type	Description
`tuple[ndarray, dict]`	The array and its metadata (an empty dict when no sidecar exists).

Raises:

Type	Description
`ValueError`	Raised when neither `npy_path` nor `folder` is given.
`FileNotFoundError`	Raised when the '.npy' does not exist, with the path that was tried.

Source code in src/dapkel/core/store.py

def load_map(
    folder: str | None = None,
    *,
    kind: str = "",
    tag: str = "",
    npy_path: str | None = None,
    file_name: str | None = None,
) -> tuple[np.ndarray, dict]:
    """Load a stage-1 array and its metadata back from ``processed/``.

    Parameters
    ----------
    folder : str | None, optional
        The data folder to look under. May be None when ``npy_path`` is
        given. The default is None.
    kind : str, optional
        Analysis name, used to derive the file name. The default is "".
    tag : str, optional
        Readout/quadrant tag. The default is "".
    npy_path : str | None, optional
        Explicit path to a saved '.npy', bypassing the derivation entirely.
        The default is None.
    file_name : str | None, optional
        Explicit file name within ``processed/``. The default is None.

    Returns
    -------
    tuple[np.ndarray, dict]
        The array and its metadata (an empty dict when no sidecar exists).

    Raises
    ------
    ValueError
        Raised when neither ``npy_path`` nor ``folder`` is given.
    FileNotFoundError
        Raised when the '.npy' does not exist, with the path that was tried.
    """
    if npy_path is None:
        if folder is None:
            raise ValueError("give either 'folder' or 'npy_path'")
        dataset = os.path.basename(os.path.normpath(folder))
        name = file_name or map_file_name(dataset, kind, tag)
        npy_path = os.path.join(folder, PROCESSED_DIR, name)

    if not os.path.isfile(npy_path):
        raise FileNotFoundError(
            f"No saved '{kind or 'array'}' found at:\n  {npy_path}\n"
            f"Run the matching compute_and_save_* first."
        )
    return np.load(npy_path), read_meta(npy_path)

save_figure ¶

save_figure(fig: Figure, results_dir: str, file_name: str) -> str

Save a figure into the results folder.

Parameters:

Name	Type	Description	Default
`fig`	`Figure`	Figure to save.	required
`results_dir`	`str`	Folder the figure should be saved into. Created if missing.	required
`file_name`	`str`	Name of the '.png' file to save the figure as.	required

Returns:

Type	Description
`str`	Path the figure was saved to.

Source code in src/dapkel/core/store.py

def save_figure(fig: plt.Figure, results_dir: str, file_name: str) -> str:
    """Save a figure into the results folder.

    Parameters
    ----------
    fig : plt.Figure
        Figure to save.
    results_dir : str
        Folder the figure should be saved into. Created if missing.
    file_name : str
        Name of the '.png' file to save the figure as.

    Returns
    -------
    str
        Path the figure was saved to.
    """
    os.makedirs(results_dir, exist_ok=True)
    out_path = os.path.join(results_dir, file_name)
    fig.savefig(out_path)
    print(f"\n> > > Plot is saved as {file_name} in {results_dir} < < <")
    return out_path

confirm_rewrite ¶

confirm_rewrite(paths: str | Sequence[str], *, delay: float | None = None) -> bool

Warn, then pause, before existing artifacts are overwritten.

Call this from any analysis whose rewrite=True is about to destroy stage-1 data. Nothing is printed and nothing is waited for on a fresh run (no target exists), so the pause only ever costs something when there is genuinely something to lose - typically a rewrite=True left in the script from the previous acquisition.

The countdown is a plain time.sleep, so Ctrl-C during it raises 'KeyboardInterrupt' out of the analysis and the data on disk is untouched.

Parameters:

Name	Type	Description	Default
`paths`	`str \| Sequence[str]`	Artifact(s) about to be overwritten. Files and directories are both accepted; entries that do not exist are ignored. Pass every target, including part-file folders, so the warning names them all.	required
`delay`	`float \| None`	Seconds to count down. The default is None, meaning 'REWRITE_DELAY_S' (5 s). Pass 0 to warn without pausing.	`None`

Returns:

Type	Description
`bool`	True if something existed and is about to be overwritten, False on a fresh run.

Source code in src/dapkel/core/store.py

def confirm_rewrite(
    paths: str | Sequence[str], *, delay: float | None = None
) -> bool:
    """Warn, then pause, before existing artifacts are overwritten.

    Call this from any analysis whose ``rewrite=True`` is about to destroy
    stage-1 data. Nothing is printed and nothing is waited for on a fresh run
    (no target exists), so the pause only ever costs something when there is
    genuinely something to lose - typically a ``rewrite=True`` left in the
    script from the previous acquisition.

    The countdown is a plain ``time.sleep``, so Ctrl-C during it raises
    'KeyboardInterrupt' out of the analysis and the data on disk is untouched.

    Parameters
    ----------
    paths : str | Sequence[str]
        Artifact(s) about to be overwritten. Files and directories are both
        accepted; entries that do not exist are ignored. Pass every target,
        including part-file folders, so the warning names them all.
    delay : float | None, optional
        Seconds to count down. The default is None, meaning
        'REWRITE_DELAY_S' (5 s). Pass 0 to warn without pausing.

    Returns
    -------
    bool
        True if something existed and is about to be overwritten, False on a
        fresh run.
    """
    targets = [paths] if isinstance(paths, str) else list(paths)
    existing = [p for p in targets if os.path.exists(p)]
    if not existing:
        return False

    wait = REWRITE_DELAY_S if delay is None else float(delay)

    print("\n" + "!" * 72)
    print("!!  rewrite=True - the following data WILL BE OVERWRITTEN:")
    for p in existing:
        print(f"!!    {_describe_target(p)}")
    if wait > 0:
        print(
            f"!!  Press Ctrl-C within {wait:.0f} s to abort and copy it "
            "somewhere safe."
        )
    print("!" * 72)

    remaining = wait
    while remaining > 0:
        print(f"  overwriting in {remaining:.0f} s ... ", end="", flush=True)
        step = min(1.0, remaining)
        time.sleep(step)
        remaining -= step
    if wait > 0:
        print("\n  overwriting now.\n")
    return True

dapkel.core¶

core.io — finding and sizing raw files¶

io ¶

frames_in_file ¶

find_bin_files ¶

core.timing — frame period and live time¶

timing ¶

to_ps ¶

to_s ¶

codes_for_full_scale ¶

resolve_tdc_full_scale_s ¶

resolve_cycle_time ¶

resolve_frame_time ¶

resolve_live_time ¶

core.pairs — which pixel pairs, and what they are called¶

pairs ¶

pair_label ¶

pair_list ¶

pair_labels ¶

core.reduce — folding frames into a map¶

reduce ¶

accumulate_frames ¶

assemble_64 ¶

core.plots — the two figure shapes¶

plots ¶

map_stats ¶

sensor_map ¶

sorted_distribution ¶

core.store — the stage 1 / stage 2 boundary¶

store ¶

processed_dir ¶

results_dir ¶

map_file_name ¶

meta_path ¶

read_meta ¶

save_map ¶

load_map ¶

save_figure ¶

confirm_rewrite ¶

`dapkel.core`¶

`core.io` — finding and sizing raw files¶

`core.timing` — frame period and live time¶

`core.pairs` — which pixel pairs, and what they are called¶

`core.reduce` — folding frames into a map¶

`core.plots` — the two figure shapes¶

`core.store` — the stage 1 / stage 2 boundary¶