Readers

High-level entry points for opening timsTOF .d acquisitions.

tdfpy.get_acquisition_type

get_acquisition_type(
    analysis_dir: str | Path,
) -> AcquisitionType

Determine the acquisition type (DDA, DIA, or PRM) of a .d folder by examining the MsMsType values in the Frames table.

Parameters:

Name	Type	Description	Default
analysis_dir	str | Path	Path to the .d folder	required

Returns:

Type	Description
AcquisitionType	AcquisitionType.DDA, .DIA or .PRM for the first MS/MS frame
AcquisitionType	type found (checked in that order), or AcquisitionType.UNKNOWN.

Raises:

Type	Description
FileNotFoundError	If analysis.tdf does not exist

Source code in src/tdfpy/reader.py

def get_acquisition_type(analysis_dir: str | Path) -> AcquisitionType:
    """
    Determine the acquisition type (DDA, DIA, or PRM) of a .d folder by
    examining the MsMsType values in the Frames table.

    Args:
        analysis_dir: Path to the .d folder

    Returns:
        ``AcquisitionType.DDA``, ``.DIA`` or ``.PRM`` for the first MS/MS frame
        type found (checked in that order), or ``AcquisitionType.UNKNOWN``.

    Raises:
        FileNotFoundError: If analysis.tdf does not exist
    """
    analysis_tdf_path = Path(analysis_dir) / "analysis.tdf"
    if not analysis_tdf_path.exists():
        raise FileNotFoundError(f"analysis.tdf not found at {analysis_tdf_path}")

    with _tdf_connection(analysis_tdf_path) as conn:
        msms_types = {int(t) for (t,) in conn.execute("SELECT DISTINCT MsMsType FROM Frames")}
    acq = _acquisition_type_of(msms_types)
    if acq is not AcquisitionType.UNKNOWN:
        return acq

    logger.warning(
        "get_acquisition_type(%s): no known MS2 MsMsType found (present: %s); returning 'unknown'. Expected one of DDA_MS2=8, DIA_MS2=9, PRM_MS2=10.",
        analysis_dir,
        sorted(msms_types),
    )
    return AcquisitionType.UNKNOWN

tdfpy.DDA

DDA(analysis_dir: str | Path)

Bases: _DFolder

Open a DDA (Data-Dependent Acquisition) .d folder.

Use as a context manager so the TimsData connection is closed when done. Exposes MS1 frames via ms1 and precursors via precursors.

Parameters:

Name	Type	Description	Default
analysis_dir	str | Path	Path to the .d folder containing analysis.tdf and analysis.tdf_bin.	required

Raises:

Type	Description
FileNotFoundError	If the .d folder or required files are missing.
AcquisitionTypeError	If the folder holds a DIA or PRM acquisition.
TdfpyError	If a frame has an MsMsType other than MS1 or DDA MS/MS.

Note

PasefFrameMsMsInfo rows with a NULL Precursor are tolerated: the PasefFrameMsmsInfo element is built with precursor_id=None and logged, rather than failing the whole file. Such rows are not reachable from any Precursor.

Example

with DDA("/path/to/data.d") as dda:
    for frame in dda.ms1:
        print(frame.frame_id, frame.rt)

Source code in src/tdfpy/reader.py

def __init__(self, analysis_dir: str | Path):
    super().__init__(analysis_dir)

    with _tdf_connection(self.analysis_tdf_path) as conn:
        frames = _Frames(conn, self._analysis_dir)
        _require_acquisition_type(frames.msms_types, AcquisitionType.DDA)
        pasef_rows = _rows(
            conn,
            "PasefFrameMsMsInfo",
            ("Frame", "ScanNumBegin", "ScanNumEnd", "IsolationMz", "IsolationWidth", "CollisionEnergy", "Precursor"),
        )
        precursor_rows = _rows(
            conn,
            "Precursors",
            ("Id", "LargestPeakMz", "AverageMz", "MonoisotopicMz", "Charge", "ScanNumber", "Intensity", "Parent"),
        )
    timsdata = self.timsdata

    pasef_msms_infos: dict[int, list[PasefFrameMsmsInfo]] = {}
    # PasefFrameMsMsInfo rows whose Precursor is NULL cannot be attached to
    # any Precursor, so they are counted and logged rather than raised on.
    unassigned: list[PasefFrameMsmsInfo] = []
    for frame_id, scan_begin, scan_end, isolation_mz, isolation_width, collision_energy, precursor_id in pasef_rows:
        frame_id = int(frame_id)
        pasef_info = PasefFrameMsmsInfo(
            _timsdata=timsdata,
            frame_id=frame_id,
            scan_num_begin=int(scan_begin),
            scan_num_end=int(scan_end),
            isolation_mz=float(isolation_mz),
            isolation_width=float(isolation_width),
            collision_energy=float(collision_energy),
            precursor_id=None if precursor_id is None else int(precursor_id),
            rt=frames.rt[frame_id],
            polarity=frames.polarity[frame_id],
        )
        if pasef_info.precursor_id is None:
            unassigned.append(pasef_info)
        else:
            pasef_msms_infos.setdefault(pasef_info.precursor_id, []).append(pasef_info)

    if unassigned:
        logger.warning(
            "%s: %d PasefFrameMsMsInfo row(s) have a NULL Precursor and are "
            "not reachable from any Precursor (e.g. frames %s). They are "
            "kept with precursor_id=None.",
            self._analysis_dir,
            len(unassigned),
            sorted({i.frame_id for i in unassigned})[:10],
        )

    precursors: dict[int, Precursor] = {}
    frame_to_precursors: dict[int, list[Precursor]] = {}
    for precursor_id, largest_peak_mz, average_mz, monoisotopic_mz, charge, scan_number, intensity, frame_id in precursor_rows:
        precursor_id, frame_id = int(precursor_id), int(frame_id)
        precursor = Precursor(
            _timsdata=timsdata,
            precursor_id=precursor_id,
            largest_peak_mz=float(largest_peak_mz),
            average_mz=float(average_mz),
            monoisotopic_mz=None if monoisotopic_mz is None else float(monoisotopic_mz),
            charge=None if charge is None else int(charge),
            scan_number=float(scan_number),
            intensity=float(intensity),
            parent_frame_id=frame_id,
            pasef_frame_msms_infos=tuple(pasef_msms_infos.get(precursor_id, [])),
            rt=frames.rt[frame_id],
        )
        precursors[precursor_id] = precursor
        frame_to_precursors.setdefault(frame_id, []).append(precursor)

    ms1_frames: dict[int, DDAMs1Frame] = {
        fields["frame_id"]: DDAMs1Frame(**fields, precursors=tuple(frame_to_precursors.get(fields["frame_id"], [])))
        for fields in frames.ms1_fields(timsdata, MsMsType.DDA_MS2)
    }

    self._precursor_lookup = PrecursorLookup(precursors)
    self._ms1_frames_lookup = Ms1FrameLookup(ms1_frames)

    logger.info(
        "Opened DDA .d folder %s: %d MS1 frames, %d precursors, %d PASEF windows.",
        self._analysis_dir,
        len(ms1_frames),
        len(precursors),
        sum(len(v) for v in pasef_msms_infos.values()),
    )

timsdata property

timsdata: TimsData

The open :class:TimsData handle on analysis.tdf_bin.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

pandas_tdf property

pandas_tdf: PandasTdf

A :class:PandasTdf view of analysis.tdf.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

metadata property

metadata: MetaData

Global metadata about the acquisition.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

calibration property

calibration: Calibration

Calibration information.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

ms1 property

ms1: Ms1FrameLookup[DDAMs1Frame]

Lookup for MS1 frames. Supports indexing by frame ID.

precursors property

precursors: PrecursorLookup

Lookup for all precursors. Supports indexing by precursor ID and .query().

close

close() -> None

Close the TimsData connection.

Source code in src/tdfpy/reader.py

def close(self) -> None:
    """Close the TimsData connection."""
    if not getattr(self, "_closed", True):
        td = getattr(self, "_timsdata", None)
        if td is not None:
            td.close()
        self._closed = True

tdfpy.DIA

DIA(analysis_dir: str | Path)

Bases: _DFolder

Open a DIA (Data-Independent Acquisition) .d folder.

Use as a context manager so the TimsData connection is closed when done. Exposes MS1 frames via ms1, per-frame windows via windows, and the window definitions via window_groups.

Parameters:

Name	Type	Description	Default
analysis_dir	str | Path	Path to the .d folder containing analysis.tdf and analysis.tdf_bin.	required

Raises:

Type	Description
FileNotFoundError	If the .d folder or required files are missing.
AcquisitionTypeError	If the folder holds a DDA or PRM acquisition.
TdfpyError	If a frame has an MsMsType other than MS1 or DIA MS/MS.

Example

with DIA("/path/to/data.d") as dia:
    for group in dia.window_groups:
        print(group.window_group_id, group.isolation_mz)
    for window in dia.windows[1]:
        print(window.frame_id, window.rt)

Source code in src/tdfpy/reader.py

def __init__(self, analysis_dir: str | Path):
    super().__init__(analysis_dir)

    with _tdf_connection(self.analysis_tdf_path) as conn:
        frames = _Frames(conn, self._analysis_dir)
        _require_acquisition_type(frames.msms_types, AcquisitionType.DIA)
        window_rows = _rows(
            conn,
            "DiaFrameMsMsWindows",
            ("WindowGroup", "ScanNumBegin", "ScanNumEnd", "IsolationMz", "IsolationWidth", "CollisionEnergy"),
        )
        frame_group_rows = _rows(conn, "DiaFrameMsMsInfo", ("Frame", "WindowGroup"))
    timsdata = self.timsdata

    window_groups: dict[int, list[DiaWindowGroup]] = {}
    for window_index, (group_id, scan_begin, scan_end, isolation_mz, isolation_width, collision_energy) in enumerate(window_rows):
        group = DiaWindowGroup(
            window_index=window_index,
            window_group_id=int(group_id),
            scan_num_begin=int(scan_begin),
            scan_num_end=int(scan_end),
            isolation_mz=float(isolation_mz),
            isolation_width=float(isolation_width),
            collision_energy=float(collision_energy),
        )
        window_groups.setdefault(group.window_group_id, []).append(group)

    # One DiaWindow per (frame, window-group row): each MS/MS frame repeats
    # every window of its group.
    all_windows: list[DiaWindow] = []
    for frame_id, group_id in frame_group_rows:
        frame_id = int(frame_id)
        for group in window_groups.get(int(group_id), []):
            window = DiaWindow(
                _timsdata=timsdata,
                frame_id=frame_id,
                window_index=group.window_index,
                window_group_id=group.window_group_id,
                scan_num_begin=group.scan_num_begin,
                scan_num_end=group.scan_num_end,
                isolation_mz=group.isolation_mz,
                isolation_width=group.isolation_width,
                collision_energy=group.collision_energy,
                rt=frames.rt[frame_id],
                polarity=frames.polarity[frame_id],
            )
            all_windows.append(window)

    ms1_frames: dict[int, DIAMs1Frame] = {fields["frame_id"]: DIAMs1Frame(**fields) for fields in frames.ms1_fields(timsdata, MsMsType.DIA_MS2)}

    self._window_groups = tuple(group for groups in window_groups.values() for group in groups)
    self._dia_windows_lookup = DiaWindowLookup(all_windows)
    self._ms1_frames_lookup = Ms1FrameLookup(ms1_frames)

    logger.info(
        "Opened DIA .d folder %s: %d MS1 frames, %d windows, %d window groups.",
        self._analysis_dir,
        len(ms1_frames),
        len(all_windows),
        len(window_groups),
    )

timsdata property

timsdata: TimsData

The open :class:TimsData handle on analysis.tdf_bin.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

pandas_tdf property

pandas_tdf: PandasTdf

A :class:PandasTdf view of analysis.tdf.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

metadata property

metadata: MetaData

Global metadata about the acquisition.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

calibration property

calibration: Calibration

Calibration information.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

ms1 property

ms1: Ms1FrameLookup[DIAMs1Frame]

Lookup for MS1 frames. Supports indexing by frame ID.

windows property

windows: DiaWindowLookup

Lookup for all per-frame DIA windows. Indexed by window group ID; supports .query().

window_groups property

window_groups: tuple[DiaWindowGroup, ...]

Every window definition (DiaFrameMsMsWindows row), grouped by window group ID.

close

close() -> None

Close the TimsData connection.

Source code in src/tdfpy/reader.py

def close(self) -> None:
    """Close the TimsData connection."""
    if not getattr(self, "_closed", True):
        td = getattr(self, "_timsdata", None)
        if td is not None:
            td.close()
        self._closed = True

tdfpy.PRM

PRM(analysis_dir: str | Path)

Bases: _DFolder

Open a PRM (Parallel Reaction Monitoring) .d folder.

Use as a context manager so the TimsData connection is closed when done. Exposes MS1 frames via ms1, PRM targets via targets, and individual transitions via transitions.

Parameters:

Name	Type	Description	Default
analysis_dir	str | Path	Path to the .d folder containing analysis.tdf and analysis.tdf_bin.	required

Raises:

Type	Description
FileNotFoundError	If the .d folder or required files are missing.
AcquisitionTypeError	If the folder holds a DDA or DIA acquisition.
TdfpyError	If a frame has an MsMsType other than MS1 or PRM MS/MS.

Example

with PRM("/path/to/data.d") as prm:
    for target in prm.targets:
        print(target.precursor_mz, target.charge)
    for transition in prm.transitions:
        print(transition.frame_id, transition.target.target_id)

Source code in src/tdfpy/reader.py

def __init__(self, analysis_dir: str | Path):
    super().__init__(analysis_dir)

    with _tdf_connection(self.analysis_tdf_path) as conn:
        frames = _Frames(conn, self._analysis_dir)
        _require_acquisition_type(frames.msms_types, AcquisitionType.PRM)
        target_rows = _rows(conn, "PrmTargets", ("Id", "ExternalId", "Time", "OneOverK0", "MonoisotopicMz", "Charge", "Description"))
        transition_rows = _rows(
            conn,
            "PrmFrameMsMsInfo",
            ("Frame", "ScanNumBegin", "ScanNumEnd", "IsolationMz", "IsolationWidth", "CollisionEnergy", "Target"),
        )
    timsdata = self.timsdata

    targets: dict[int, PrmTarget] = {}
    for target_id, external_id, rt, ook0, mz, charge, description in target_rows:
        target_id = int(target_id)
        targets[target_id] = PrmTarget(
            target_id=target_id,
            external_id=None if external_id is None else str(external_id),
            rt=float(rt),
            ook0=float(ook0),
            precursor_mz=float(mz),
            charge=int(charge),
            description="" if description is None else str(description),
        )

    all_transitions: list[PrmTransition] = []
    target_to_transitions: dict[int, list[PrmTransition]] = {}
    for frame_id, scan_begin, scan_end, isolation_mz, isolation_width, collision_energy, target_id in transition_rows:
        frame_id, target_id = int(frame_id), int(target_id)
        transition = PrmTransition(
            _timsdata=timsdata,
            frame_id=frame_id,
            scan_num_begin=int(scan_begin),
            scan_num_end=int(scan_end),
            isolation_mz=float(isolation_mz),
            isolation_width=float(isolation_width),
            collision_energy=float(collision_energy),
            target=targets[target_id],
            rt=frames.rt[frame_id],
            polarity=frames.polarity[frame_id],
        )
        all_transitions.append(transition)
        target_to_transitions.setdefault(target_id, []).append(transition)

    # Targets are frozen; the back-reference is set once, here, before any
    # target leaves the reader. `transitions` is excluded from eq/hash.
    for target_id, target in targets.items():
        object.__setattr__(target, "transitions", tuple(target_to_transitions.get(target_id, [])))

    ms1_frames: dict[int, PRMMs1Frame] = {fields["frame_id"]: PRMMs1Frame(**fields) for fields in frames.ms1_fields(timsdata, MsMsType.PRM_MS2)}

    self._prm_target_lookup = PrmTargetLookup(targets)
    self._prm_transition_lookup = PrmTransitionLookup(all_transitions)
    self._ms1_frames_lookup = Ms1FrameLookup(ms1_frames)

    logger.info(
        "Opened PRM .d folder %s: %d MS1 frames, %d targets, %d transitions.",
        self._analysis_dir,
        len(ms1_frames),
        len(targets),
        len(all_transitions),
    )

timsdata property

timsdata: TimsData

The open :class:TimsData handle on analysis.tdf_bin.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

pandas_tdf property

pandas_tdf: PandasTdf

A :class:PandasTdf view of analysis.tdf.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

metadata property

metadata: MetaData

Global metadata about the acquisition.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

calibration property

calibration: Calibration

Calibration information.

Raises:

Type	Description
ReaderClosedError	If the reader was closed.

ms1 property

ms1: Ms1FrameLookup[PRMMs1Frame]

Lookup for MS1 frames. Supports indexing by frame ID.

targets property

targets: PrmTargetLookup

Lookup for all PRM targets. Supports indexing by target ID and .query().

transitions property

transitions: PrmTransitionLookup

Lookup for all PRM transitions. Supports indexing by target ID and .query().

close

close() -> None

Close the TimsData connection.

Source code in src/tdfpy/reader.py

def close(self) -> None:
    """Close the TimsData connection."""
    if not getattr(self, "_closed", True):
        td = getattr(self, "_timsdata", None)
        if td is not None:
            td.close()
        self._closed = True