PandasTDF

Low-level classes for direct access to the .tdf SQLite database and the Bruker TimsData C library. Prefer the high-level DDA/DIA API unless you need raw frame or scan data.

tdfpy.PandasTdf dataclass

PandasTdf(db_path: str | Path)

A class for working with TDF (Bruker Data File) using pandas DataFrames.

calibration_info property

calibration_info: pd.DataFrame

The 'CalibrationInfo' table as a pandas DataFrame. :return: table as a pandas DataFrame

dia_frame_msms_info property

dia_frame_msms_info: pd.DataFrame

The 'DiaFrameMsMsInfo' table as a pandas DataFrame. :return: table as a pandas DataFrame

dia_frame_msms_window_groups property

dia_frame_msms_window_groups: pd.DataFrame

The 'DiaFrameMsMsWindowGroups' table as a pandas DataFrame. :return: table as a pandas DataFrame

dia_frame_msms_windows property

dia_frame_msms_windows: pd.DataFrame

The 'DiaFrameMsMsWindows' table as a pandas DataFrame. :return: table as a pandas DataFrame

error_log property

error_log: pd.DataFrame

The 'ErrorLog' table as a pandas DataFrame. :return: table as a pandas DataFrame

frame_msms_info property

frame_msms_info: pd.DataFrame

The 'FrameMsMsInfo' table as a pandas DataFrame. :return: table as a pandas DataFrame

frame_properties property

frame_properties: pd.DataFrame

The 'FrameProperties' table as a pandas DataFrame. :return: table as a pandas DataFrame

frames property

frames: pd.DataFrame

The 'Frames' table as a pandas DataFrame. :return: table as a pandas DataFrame

global_metadata property

global_metadata: pd.DataFrame

The 'GlobalMetadata' table as a pandas DataFrame. :return: table as a pandas DataFrame

group_properties property

group_properties: pd.DataFrame

The 'GroupProperties' table as a pandas DataFrame. :return: table as a pandas DataFrame

mz_calibration property

mz_calibration: pd.DataFrame

The 'MzCalibration' table as a pandas DataFrame. :return: table as a pandas DataFrame

pasef_frame_msms_info property

pasef_frame_msms_info: pd.DataFrame

The 'PasefFrameMsMsInfo' table as a pandas DataFrame. :return: table as a pandas DataFrame

precursors property

precursors: pd.DataFrame

The 'Precursors' table as a pandas DataFrame. :return: table as a pandas DataFrame

properties property

properties: pd.DataFrame

The 'Properties' table as a pandas DataFrame. :return: table as a pandas DataFrame

property_definitions property

property_definitions: pd.DataFrame

The 'PropertyDefinitions' table as a pandas DataFrame. :return: table as a pandas DataFrame

property_groups property

property_groups: pd.DataFrame

The 'PropertyGroups' table as a pandas DataFrame. :return: table as a pandas DataFrame

segments property

segments: pd.DataFrame

The 'Segments' table as a pandas DataFrame. :return: table as a pandas DataFrame

tims_calibration property

tims_calibration: pd.DataFrame

The 'TimsCalibration' table as a pandas DataFrame. :return: table as a pandas DataFrame

prm_frame_measurement_mode property

prm_frame_measurement_mode: pd.DataFrame

The 'PrmFrameMeasurementMode' table as a pandas DataFrame. :return: table as a pandas DataFrame

prm_frame_msms_info property

prm_frame_msms_info: pd.DataFrame

The 'PrmFrameMsMsInfo' table as a pandas DataFrame. :return: table as a pandas DataFrame

prm_targets property

prm_targets: pd.DataFrame

The 'PrmTargets' table as a pandas DataFrame. :return: table as a pandas DataFrame

is_dda property

is_dda: bool

Checks if the database contains DDA (Data-Dependent Acquisition) data.

Returns:

Name	Type	Description
bool	bool	True if DDA data is present, False otherwise.

is_prm property

is_prm: bool

Checks if the database contains PRM (Parallel Reaction Monitoring) data.

Returns:

Name	Type	Description
bool	bool	True if PRM data is present, False otherwise.

is_dia property

is_dia: bool

Checks if the database contains DIA (Data-Independent Acquisition) data.

Returns:

Name	Type	Description
bool	bool	True if DIA data is present, False otherwise.

is_maldi property

is_maldi: bool

Checks if the database contains MALDI (Matrix-Assisted Laser Desorption/Ionization) data. Not supported in tdfpy, but this method can be used to check for MALDI data if it is added in the future.

Returns:

Name	Type	Description
bool	bool	True if MALDI data is present, False otherwise.

get_table_names

get_table_names() -> list[str]

Retrieves the names of all tables in the SQLite database.

Returns:

Type	Description
list[str]	list[str]: A list of table names in the database.

Source code in src/tdfpy/tdf.py

def get_table_names(self) -> list[str]:
    """
    Retrieves the names of all tables in the SQLite database.

    Returns:
        list[str]: A list of table names in the database.
    """
    # closing(), not the connection's own context manager, which only ends
    # the transaction and would leave the handle open until GC.
    with closing(sqlite3.connect(str(self.db_path))) as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
        table_names = [table[0] for table in cursor.fetchall()]
    return table_names

tdfpy.TimsData

TimsData(
    analysis_directory: str | os.PathLike[str],
    use_recalibrated_state: bool = False,
    pressure_compensation_strategy: PressureCompensationStrategy = PressureCompensationStrategy.NoPressureCompensation,
)

Random-access reader for a Bruker .d folder.

Metadata is loaded eagerly on open; spectral data is read from analysis.tdf_bin on demand.

Reading frames from several threads through one open reader is safe: frame bytes are fetched with :func:os.pread, which takes its offset as an argument and so shares no file position between threads, and decompression goes through stateless one-shot entry points. Where os.pread is unavailable (Windows) the seek + read pair is serialised by a lock instead. close() is not safe to race against an in-flight read, and the sqlite3 connection on :attr:conn keeps sqlite3's own thread rules.

Source code in src/tdfpy/timsdata.py

def __init__(
    self,
    analysis_directory: str | os.PathLike[str],
    use_recalibrated_state: bool = False,
    pressure_compensation_strategy: PressureCompensationStrategy = PressureCompensationStrategy.NoPressureCompensation,
) -> None:
    analysis_directory = str(analysis_directory)

    if use_recalibrated_state:
        raise UnsupportedTdfError(
            "use_recalibrated_state=True is not supported; tdfpy reads the "
            "calibration recorded in analysis.tdf."
        )
    if (
        pressure_compensation_strategy
        is not PressureCompensationStrategy.NoPressureCompensation
    ):
        raise UnsupportedTdfError(
            f"{pressure_compensation_strategy.name} is not supported; only "
            "NoPressureCompensation is implemented."
        )

    if not os.path.isdir(analysis_directory):
        raise FileNotFoundError(
            f"Analysis directory not found: {analysis_directory!r}"
        )
    tdf_path = os.path.join(analysis_directory, "analysis.tdf")
    bin_path = os.path.join(analysis_directory, "analysis.tdf_bin")
    for path in (tdf_path, bin_path):
        if not os.path.exists(path):
            raise FileNotFoundError(
                f"{os.path.basename(path)} not found in {analysis_directory!r}"
            )

    self.analysis_directory = analysis_directory
    self.conn: sqlite3.Connection | None = sqlite3.connect(
        Path(tdf_path).resolve().as_uri() + "?mode=ro", uri=True
    )
    self.conn.row_factory = sqlite3.Row

    try:
        self._load_metadata()
    except Exception:
        self.conn.close()
        self.conn = None
        raise

    #: Serialises seek + read on the shared handle for the no-``pread``
    #: fallback. Unused, but still created, when ``os.pread`` exists.
    self._read_lock = threading.Lock()
    #: Open binary file object, or ``None`` once :meth:`close` has run.
    #: Callers use this only to test whether the reader is still open.
    self.handle: Any = open(bin_path, "rb")
    self._fd: int = self.handle.fileno()

frame_ids property

frame_ids: tuple[int, ...]

Frame IDs in acquisition ID order. Requires an open reader.

frame_metadata

frame_metadata(frame_id: int) -> FrameMetadata

Read eagerly loaded metadata without accessing SQLite.

Source code in src/tdfpy/timsdata.py

def frame_metadata(self, frame_id: int) -> FrameMetadata:
    """Read eagerly loaded metadata without accessing SQLite."""
    self._require_open()
    self._frame(frame_id)
    return self._frame_metadata[frame_id]

metadata_table

metadata_table(name: str) -> tuple[sqlite3.Row, ...]

Read an immutable snapshot of a gate metadata table.

Supported names are PropertyDefinitions, GroupProperties, and DiaFrameMsMsWindows. An absent table produces an empty tuple.

Source code in src/tdfpy/timsdata.py

def metadata_table(self, name: str) -> tuple[sqlite3.Row, ...]:
    """Read an immutable snapshot of a gate metadata table.

    Supported names are PropertyDefinitions, GroupProperties, and
    DiaFrameMsMsWindows. An absent table produces an empty tuple.
    """
    self._require_open()
    return self._metadata_tables[name]

mz_calibration_key

mz_calibration_key(frame_id: int) -> tuple[float, ...]

Identify the effective m/z conversion, including temperature drift.

Source code in src/tdfpy/timsdata.py

def mz_calibration_key(self, frame_id: int) -> tuple[float, ...]:
    """Identify the effective m/z conversion, including temperature drift."""
    self._require_open()
    cal, t1, t2 = self._mz_cal(frame_id)
    return (
        cal.digitizer_timebase,
        cal.digitizer_delay,
        cal.c0,
        cal._c1_at(t1, t2),
        cal.c2,
    )

calibration_key

calibration_key(frame_id: int) -> tuple

Identify the effective m/z and mobility conversions for caching.

Source code in src/tdfpy/timsdata.py

def calibration_key(self, frame_id: int) -> tuple:
    """Identify the effective m/z and mobility conversions for caching."""
    return (self.mz_calibration_key(frame_id), self._tims_cal(frame_id))

indexToMz

indexToMz(
    frame_id: int, indices: npt.ArrayLike
) -> npt.NDArray[np.float64]

Convert TOF sample indices to m/z for frame_id.

Source code in src/tdfpy/timsdata.py

def indexToMz(
    self, frame_id: int, indices: npt.ArrayLike
) -> npt.NDArray[np.float64]:
    """Convert TOF sample indices to m/z for ``frame_id``."""
    cal, t1, t2 = self._mz_cal(frame_id)
    return cal.index_to_mz(indices, t1, t2)

mzToIndex

mzToIndex(
    frame_id: int, mzs: npt.ArrayLike
) -> npt.NDArray[np.float64]

Convert m/z to (fractional) TOF sample indices for frame_id.

Source code in src/tdfpy/timsdata.py

def mzToIndex(self, frame_id: int, mzs: npt.ArrayLike) -> npt.NDArray[np.float64]:
    """Convert m/z to (fractional) TOF sample indices for ``frame_id``."""
    cal, t1, t2 = self._mz_cal(frame_id)
    return cal.mz_to_index(mzs, t1, t2)

scanNumToOneOverK0

scanNumToOneOverK0(
    frame_id: int, scan_nums: npt.ArrayLike
) -> npt.NDArray[np.float64]

Convert scan numbers to inverse reduced mobility (1/K0).

Source code in src/tdfpy/timsdata.py

def scanNumToOneOverK0(
    self, frame_id: int, scan_nums: npt.ArrayLike
) -> npt.NDArray[np.float64]:
    """Convert scan numbers to inverse reduced mobility (1/K0)."""
    return self._tims_cal(frame_id).scan_to_one_over_k0(scan_nums)

oneOverK0ToScanNum

oneOverK0ToScanNum(
    frame_id: int, mobilities: npt.ArrayLike
) -> npt.NDArray[np.float64]

Convert 1/K0 to (fractional) scan numbers.

Source code in src/tdfpy/timsdata.py

def oneOverK0ToScanNum(
    self, frame_id: int, mobilities: npt.ArrayLike
) -> npt.NDArray[np.float64]:
    """Convert 1/K0 to (fractional) scan numbers."""
    return self._tims_cal(frame_id).one_over_k0_to_scan(mobilities)

scanNumToVoltage

scanNumToVoltage(
    frame_id: int, scan_nums: npt.ArrayLike
) -> npt.NDArray[np.float64]

Convert scan numbers to TIMS ramp voltage.

Source code in src/tdfpy/timsdata.py

def scanNumToVoltage(
    self, frame_id: int, scan_nums: npt.ArrayLike
) -> npt.NDArray[np.float64]:
    """Convert scan numbers to TIMS ramp voltage."""
    return self._tims_cal(frame_id).scan_to_voltage(scan_nums)

voltageToScanNum

voltageToScanNum(
    frame_id: int, voltages: npt.ArrayLike
) -> npt.NDArray[np.float64]

Convert TIMS ramp voltage to (fractional) scan numbers.

Source code in src/tdfpy/timsdata.py

def voltageToScanNum(
    self, frame_id: int, voltages: npt.ArrayLike
) -> npt.NDArray[np.float64]:
    """Convert TIMS ramp voltage to (fractional) scan numbers."""
    return self._tims_cal(frame_id).voltage_to_scan(voltages)

read_frame_arrays

read_frame_arrays(
    frame_id: int,
    scan_begin: int = 0,
    scan_end: int | None = None,
) -> tuple[
    npt.NDArray[np.int64],
    npt.NDArray[np.uint32],
    npt.NDArray[np.uint32],
]

Read scans [scan_begin, scan_end) as three flat, parallel arrays.

Returns (scan_indices, tof_indices, intensities), one entry per peak. This is the cheap path: peaks for a contiguous scan range are already contiguous in the decoded frame, so it slices rather than splitting the frame into per-scan arrays the way :meth:readScans must.

Prefer this whenever you were going to concatenate readScans output back together.

Source code in src/tdfpy/timsdata.py

def read_frame_arrays(
    self, frame_id: int, scan_begin: int = 0, scan_end: int | None = None
) -> tuple[npt.NDArray[np.int64], npt.NDArray[np.uint32], npt.NDArray[np.uint32]]:
    """Read scans ``[scan_begin, scan_end)`` as three flat, parallel arrays.

    Returns ``(scan_indices, tof_indices, intensities)``, one entry per peak.
    This is the cheap path: peaks for a contiguous scan range are already
    contiguous in the decoded frame, so it slices rather than splitting the
    frame into per-scan arrays the way :meth:`readScans` must.

    Prefer this whenever you were going to concatenate ``readScans`` output
    back together.
    """
    decoded = self._decode(frame_id)
    if decoded is None:
        return (
            np.zeros(0, dtype=np.int64),
            _EMPTY_U32,
            _EMPTY_U32,
        )
    scan_count, starts, counts, tof, intensity = decoded

    if scan_end is None:
        scan_end = scan_count
    begin = max(0, min(int(scan_begin), scan_count))
    end = max(begin, min(int(scan_end), scan_count))
    if begin == end:
        return np.zeros(0, dtype=np.int64), _EMPTY_U32, _EMPTY_U32

    lo = int(starts[begin])
    hi = int(starts[end - 1] + counts[end - 1])
    scan_indices = np.repeat(
        np.arange(begin, end, dtype=np.int64), counts[begin:end]
    )
    return scan_indices, tof[lo:hi], intensity[lo:hi]

readScans

readScans(
    frame_id: int, scan_begin: int, scan_end: int
) -> list[
    tuple[npt.NDArray[np.uint32], npt.NDArray[np.uint32]]
]

Read scans [scan_begin, scan_end) of a frame.

Returns one (tof_indices, intensities) pair per scan. Intensities are normalised to a 100 ms accumulation window, matching Bruker.

See :meth:read_frame_arrays for a flat-array alternative that avoids materialising one array pair per scan.

Source code in src/tdfpy/timsdata.py

def readScans(
    self, frame_id: int, scan_begin: int, scan_end: int
) -> list[tuple[npt.NDArray[np.uint32], npt.NDArray[np.uint32]]]:
    """Read scans ``[scan_begin, scan_end)`` of a frame.

    Returns one ``(tof_indices, intensities)`` pair per scan. Intensities are
    normalised to a 100 ms accumulation window, matching Bruker.

    See :meth:`read_frame_arrays` for a flat-array alternative that avoids
    materialising one array pair per scan.
    """
    decoded = self._decode(frame_id)
    if decoded is None:
        return [(_EMPTY_U32, _EMPTY_U32) for _ in range(scan_begin, scan_end)]
    scan_count, starts, counts, tof, intensity = decoded

    result = []
    for i in range(scan_begin, scan_end):
        if i < 0 or i >= scan_count:
            result.append((_EMPTY_U32, _EMPTY_U32))
            continue
        start = int(starts[i])
        stop = start + int(counts[i])
        result.append((tof[start:stop], intensity[start:stop]))
    return result

tdfpy.timsdata_connect

timsdata_connect(
    analysis_dir: str | os.PathLike[str],
) -> Iterator[TimsData]

Open a :class:TimsData and close it on exit.

Source code in src/tdfpy/timsdata.py

@contextmanager
def timsdata_connect(analysis_dir: str | os.PathLike[str]) -> Iterator[TimsData]:
    """Open a :class:`TimsData` and close it on exit."""
    td: TimsData | None = None
    try:
        td = TimsData(str(analysis_dir))
        yield td
    finally:
        if td:
            td.close()