Skip to content

Utilities

nplinker.metabolomics.utils

add_annotation_to_spectrum 

add_annotation_to_spectrum(
    annotations: Mapping[str, dict],
    spectra: Sequence[Spectrum],
) -> None

Add annotations to the Spectrum.gnps_annotations attribute for input spectra.

It is possible that some spectra don't have annotations.

Note

The input spectra list is changed in place.

Parameters:

  • annotations (Mapping[str, dict]) –

    A dictionary of GNPS annotations, where the keys are spectrum ids and the values are GNPS annotations.

  • spectra (Sequence[Spectrum]) –

    A list of Spectrum objects.

Source code in src/nplinker/metabolomics/utils.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def add_annotation_to_spectrum(
    annotations: Mapping[str, dict], spectra: Sequence[Spectrum]
) -> None:
    """Add annotations to the `Spectrum.gnps_annotations` attribute for input spectra.

    It is possible that some spectra don't have annotations.

    !!! note
        The input `spectra` list is changed in place.

    Args:
        annotations: A dictionary of GNPS annotations, where the keys are
            spectrum ids and the values are GNPS annotations.
        spectra: A list of Spectrum objects.
    """
    for spec in spectra:
        if spec.id in annotations:
            spec.gnps_annotations = annotations[spec.id]

add_strains_to_spectrum 

add_strains_to_spectrum(
    strains: StrainCollection, spectra: Sequence[Spectrum]
) -> tuple[list[Spectrum], list[Spectrum]]

Add Strain objects to the Spectrum.strains attribute for input spectra.

Note

The input spectra list is changed in place.

Parameters:

Returns:

  • tuple[list[Spectrum], list[Spectrum]] –

    A tuple of two lists of Spectrum objects,

    • the first list contains Spectrum objects that are updated with Strain objects;
    • the second list contains Spectrum objects that are not updated with Strain objects because no Strain objects are found.
Source code in src/nplinker/metabolomics/utils.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def add_strains_to_spectrum(
    strains: StrainCollection, spectra: Sequence[Spectrum]
) -> tuple[list[Spectrum], list[Spectrum]]:
    """Add `Strain` objects to the `Spectrum.strains` attribute for input spectra.

    !!! note
        The input `spectra` list is changed in place.

    Args:
        strains: A collection of strain objects.
        spectra: A list of Spectrum objects.

    Returns:
        A tuple of two lists of Spectrum objects,

            - the first list contains Spectrum objects that are updated with Strain objects;
            - the second list contains Spectrum objects that are not updated with Strain objects
            because no Strain objects are found.
    """
    spectra_with_strains = []
    spectra_without_strains = []
    for spec in spectra:
        try:
            strain_list = strains.lookup(spec.id)
        except ValueError:
            spectra_without_strains.append(spec)
            continue

        for strain in strain_list:
            spec.strains.add(strain)
        spectra_with_strains.append(spec)

    logger.info(
        f"{len(spectra_with_strains)} Spectrum objects updated with Strain objects.\n"
        f"{len(spectra_without_strains)} Spectrum objects not updated with Strain objects."
    )

    return spectra_with_strains, spectra_without_strains

add_spectrum_to_mf 

add_spectrum_to_mf(
    spectra: Sequence[Spectrum],
    mfs: Sequence[MolecularFamily],
) -> tuple[
    list[MolecularFamily],
    list[MolecularFamily],
    dict[MolecularFamily, set[str]],
]

Add Spectrum objects to MolecularFamily objects.

The attribute MolecularFamily.spectra_ids contains the ids of Spectrum objects. These ids are used to find Spectrum objects from the input spectra list. The found Spectrum objects are added to the MolecularFamily.spectra attribute.

It is possible that some spectrum ids are not found in the input spectra list, and so their Spectrum objects are missing in the MolecularFamily object.

Note

The input mfs list is changed in place.

Parameters:

Returns:

  • tuple[list[MolecularFamily], list[MolecularFamily], dict[MolecularFamily, set[str]]] –

    A tuple of three elements,

    • the first list contains MolecularFamily objects that are updated with Spectrum objects
    • the second list contains MolecularFamily objects that are not updated with Spectrum objects (all Spectrum objects are missing).
    • the third is a dictionary containing MolecularFamily objects as keys and a set of ids of missing Spectrum objects as values.
Source code in src/nplinker/metabolomics/utils.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def add_spectrum_to_mf(
    spectra: Sequence[Spectrum], mfs: Sequence[MolecularFamily]
) -> tuple[list[MolecularFamily], list[MolecularFamily], dict[MolecularFamily, set[str]]]:
    """Add Spectrum objects to MolecularFamily objects.

    The attribute `MolecularFamily.spectra_ids` contains the ids of `Spectrum` objects.
    These ids are used to find `Spectrum` objects from the input `spectra` list. The found `Spectrum`
    objects are added to the `MolecularFamily.spectra` attribute.

    It is possible that some spectrum ids are not found in the input `spectra` list, and so their
    `Spectrum` objects are missing in the `MolecularFamily` object.


    !!! note
        The input `mfs` list is changed in place.

    Args:
        spectra: A list of Spectrum objects.
        mfs: A list of MolecularFamily objects.

    Returns:
        A tuple of three elements,

            - the first list contains `MolecularFamily` objects that are updated with `Spectrum` objects
            - the second list contains `MolecularFamily` objects that are not updated with `Spectrum`
            objects (all `Spectrum` objects are missing).
            - the third is a dictionary containing `MolecularFamily` objects as keys and a set of ids
            of missing `Spectrum` objects as values.
    """
    spec_dict = {spec.id: spec for spec in spectra}
    mf_with_spec = []
    mf_without_spec = []
    mf_missing_spec: dict[MolecularFamily, set[str]] = {}
    for mf in mfs:
        for spec_id in mf.spectra_ids:
            try:
                spec = spec_dict[spec_id]
            except KeyError:
                if mf not in mf_missing_spec:
                    mf_missing_spec[mf] = {spec_id}
                else:
                    mf_missing_spec[mf].add(spec_id)
                continue
            mf.add_spectrum(spec)

        if mf.spectra:
            mf_with_spec.append(mf)
        else:
            mf_without_spec.append(mf)

    logger.info(
        f"{len(mf_with_spec)} MolecularFamily objects updated with Spectrum objects.\n"
        f"{len(mf_without_spec)} MolecularFamily objects not updated with Spectrum objects.\n"
        f"{len(mf_missing_spec)} MolecularFamily objects have missing Spectrum objects."
    )
    return mf_with_spec, mf_without_spec, mf_missing_spec

extract_mappings_strain_id_ms_filename 

extract_mappings_strain_id_ms_filename(
    podp_project_json_file: str | PathLike,
) -> dict[str, set[str]]

Extract mappings "strain_id <-> MS_filename".

Parameters:

  • podp_project_json_file (str | PathLike) –

    The path to the PODP project JSON file.

Returns:

  • dict[str, set[str]] –

    Key is strain id and value is a set of MS filenames.

Notes

The podp_project_json_file is the project JSON file downloaded from PODP platform. For example, for project MSV000079284, its json file is https://pairedomicsdata.bioinformatics.nl/api/projects/4b29ddc3-26d0-40d7-80c5-44fb6631dbf9.4.

See Also
Source code in src/nplinker/metabolomics/utils.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def extract_mappings_strain_id_ms_filename(
    podp_project_json_file: str | PathLike,
) -> dict[str, set[str]]:
    """Extract mappings "strain_id <-> MS_filename".

    Args:
        podp_project_json_file: The path to the PODP project JSON file.

    Returns:
        Key is strain id and value is a set of MS filenames.

    Notes:
        The `podp_project_json_file` is the project JSON file downloaded from
        PODP platform. For example, for project MSV000079284, its json file is
        https://pairedomicsdata.bioinformatics.nl/api/projects/4b29ddc3-26d0-40d7-80c5-44fb6631dbf9.4.

    See Also:
        - [podp_generate_strain_mappings][nplinker.strain.utils.podp_generate_strain_mappings]:
            Generate strain mappings JSON file for PODP pipeline.
    """
    mappings_dict: dict[str, set[str]] = {}
    with open(podp_project_json_file, "r") as f:
        json_data = json.load(f)

    validate_podp_json(json_data)

    # Extract mappings strain id <-> metabolomics filename
    for record in json_data["genome_metabolome_links"]:
        strain_id = record["genome_label"]
        # get the actual filename of the mzXML URL
        filename = Path(record["metabolomics_file"]).name
        if strain_id in mappings_dict:
            mappings_dict[strain_id].add(filename)
        else:
            mappings_dict[strain_id] = {filename}
    return mappings_dict

extract_mappings_ms_filename_spectrum_id 

extract_mappings_ms_filename_spectrum_id(
    gnps_file_mappings_file: str | PathLike,
) -> dict[str, set[str]]

Extract mappings "MS_filename <-> spectrum_id".

Parameters:

  • gnps_file_mappings_file (str | PathLike) –

    The path to the GNPS file mappings file (csv or tsv).

Returns:

  • dict[str, set[str]] –

    Key is MS filename and value is a set of spectrum ids.

Notes

The gnps_file_mappings_file is downloaded from GNPS website and named as GNPS_FILE_MAPPINGS_TSV or GNPS_FILE_MAPPINGS_CSV. For more details, see GNPS data.

See Also
Source code in src/nplinker/metabolomics/utils.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def extract_mappings_ms_filename_spectrum_id(
    gnps_file_mappings_file: str | PathLike,
) -> dict[str, set[str]]:
    """Extract mappings "MS_filename <-> spectrum_id".

    Args:
        gnps_file_mappings_file: The path to the GNPS file mappings file (csv or tsv).

    Returns:
        Key is MS filename and value is a set of spectrum ids.

    Notes:
        The `gnps_file_mappings_file` is downloaded from GNPS website and named as
        [GNPS_FILE_MAPPINGS_TSV][nplinker.defaults.GNPS_FILE_MAPPINGS_TSV] or
        [GNPS_FILE_MAPPINGS_CSV][nplinker.defaults.GNPS_FILE_MAPPINGS_CSV].
        For more details, see [GNPS data][gnps-data].

    See Also:
        - [GNPSFileMappingLoader][nplinker.metabolomics.gnps.gnps_file_mapping_loader.GNPSFileMappingLoader]:
        Load GNPS file mappings file.
        - [podp_generate_strain_mappings][nplinker.strain.utils.podp_generate_strain_mappings]:
            Generate strain mappings JSON file for PODP pipeline.
    """
    loader = GNPSFileMappingLoader(gnps_file_mappings_file)
    return loader.mapping_reversed

get_mappings_strain_id_spectrum_id 

get_mappings_strain_id_spectrum_id(
    mappings_strain_id_ms_filename: Mapping[str, set[str]],
    mappings_ms_filename_spectrum_id: Mapping[
        str, set[str]
    ],
) -> dict[str, set[str]]

Get mappings "strain_id <-> spectrum_id".

Parameters:

  • mappings_strain_id_ms_filename (Mapping[str, set[str]]) –

    Mappings "strain_id <-> MS_filename".

  • mappings_ms_filename_spectrum_id (Mapping[str, set[str]]) –

    Mappings "MS_filename <-> spectrum_id".

Returns:

  • dict[str, set[str]] –

    Key is strain id and value is a set of spectrum ids.

See Also
  • extract_mappings_strain_id_ms_filename: Extract mappings "strain_id <-> MS_filename".
  • extract_mappings_ms_filename_spectrum_id: Extract mappings "MS_filename <-> spectrum_id".
  • podp_generate_strain_mappings: Generate strain mappings JSON file for PODP pipeline.
Source code in src/nplinker/metabolomics/utils.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def get_mappings_strain_id_spectrum_id(
    mappings_strain_id_ms_filename: Mapping[str, set[str]],
    mappings_ms_filename_spectrum_id: Mapping[str, set[str]],
) -> dict[str, set[str]]:
    """Get mappings "strain_id <-> spectrum_id".

    Args:
        mappings_strain_id_ms_filename: Mappings
            "strain_id <-> MS_filename".
        mappings_ms_filename_spectrum_id: Mappings
            "MS_filename <-> spectrum_id".

    Returns:
        Key is strain id and value is a set of spectrum ids.


    See Also:
        - `extract_mappings_strain_id_ms_filename`: Extract mappings "strain_id <-> MS_filename".
        - `extract_mappings_ms_filename_spectrum_id`: Extract mappings "MS_filename <-> spectrum_id".
        - [podp_generate_strain_mappings][nplinker.strain.utils.podp_generate_strain_mappings]:
            Generate strain mappings JSON file for PODP pipeline.
    """
    mappings_dict = {}
    for strain_id, ms_filenames in mappings_strain_id_ms_filename.items():
        spectrum_ids = set()
        for ms_filename in ms_filenames:
            if (sid := mappings_ms_filename_spectrum_id.get(ms_filename)) is not None:
                spectrum_ids.update(sid)
        if spectrum_ids:
            mappings_dict[strain_id] = spectrum_ids
    return mappings_dict