diff --git a/README.rst b/README.rst index cfcdf1b..4056441 100644 --- a/README.rst +++ b/README.rst @@ -39,11 +39,14 @@ Features * feature scaling by gauss-, gamma- and exponentially-modified gaussian distributions * m/z-and intensity noise injection ( uniform noise or a noise model that combines general noise with intensity-specific noise) * MS2 fragmentation for peptides and modified nucleosides. +* optional signing of the simulated mzML with `mzprov`_ (``write_mzml(..., sign_with_mzprov=True)``, ``pip install smiter[mzprov]``), declaring it simulated and making later edits detectable * Free software: MIT license * Documentation: https://smiter.readthedocs.io. .. _pyQms: https://github.com/pyQms/pyqms +.. _mzprov: + https://github.com/mzprov/mzprov Download and Installation ------------------------- diff --git a/setup.py b/setup.py index b041ad5..0e2074b 100644 --- a/setup.py +++ b/setup.py @@ -42,6 +42,7 @@ ], }, install_requires=requirements, + extras_require={"mzprov": ["mzprov>=0.1.2"]}, license="MIT license", long_description=readme + "\n\n" + history, include_package_data=True, diff --git a/smiter/synthetic_mzml.py b/smiter/synthetic_mzml.py index 919c63c..3818da8 100644 --- a/smiter/synthetic_mzml.py +++ b/smiter/synthetic_mzml.py @@ -1,6 +1,8 @@ """Main module.""" +import hashlib import io +import json import pathlib import time import warnings @@ -135,6 +137,7 @@ def write_mzml( fragmentor: AbstractFragmentor, noise_injector: AbstractNoiseInjector, mzml_params: Dict[str, Union[int, float, str]], + sign_with_mzprov: bool = False, ) -> str: """Write mzML file with chromatographic peaks and fragment spectra for the given molecules. @@ -143,6 +146,8 @@ def write_mzml( molecules (List[str]): Description fragmentation_function (Callable[[str], List[Tuple[float, float]]], optional): Description peak_properties (Dict[str, dict], optional): Description + sign_with_mzprov (bool, optional): Sign the mzML with an mzprov provenance + sidecar declaring it SMITER-simulated; see sign_mzml_with_mzprov(). """ # check params and raise Exception(s) if necessary logger.info("Start generating mzML") @@ -185,9 +190,61 @@ def write_mzml( path = pathlib.Path(file_path) summary_path = path.parent.resolve() / "molecule_summary.csv" peak_properties_to_csv(peak_properties, summary_path) + if sign_with_mzprov: + if not isinstance(file, str): + file.flush() + sign_mzml_with_mzprov(path, mzml_params, summary_path) return filename +def sign_mzml_with_mzprov( + mzml_path: Union[str, pathlib.Path], + mzml_params: Dict[str, Union[int, float, str]], + summary_path: Union[str, pathlib.Path], +) -> pathlib.Path: + """Sign a SMITER mzML with mzprov (https://github.com/mzprov/mzprov). + + Writes ``.smiter_params.json`` beside the mzML, holding the simulation + parameters and the SHA-256 of the molecule summary (the ground truth), and + signs the mzML with that file as its configuration. The resulting + ``.provenance.json`` declares the file SMITER-simulated. mzprov keeps + the signed copy of the parameters as ``.config.toml``, and + ``mzprov verify `` detects any later change to the spectra or to that + copy. The ground truth is bound through the hash recorded in the parameters. + The first signature generates a local signing key. + + Requires the optional dependency: ``pip install smiter[mzprov]``. + + Returns: + pathlib.Path: path of the provenance sidecar + """ + try: + from mzprov import sign_mzml_output + except ImportError as e: + raise ImportError( + "sign_with_mzprov needs mzprov: pip install smiter[mzprov]" + ) from e + mzml_path = pathlib.Path(mzml_path) + summary_path = pathlib.Path(summary_path) + config = { + "mzml_params": mzml_params, + "molecule_summary_sha256": hashlib.sha256( + summary_path.read_bytes() + ).hexdigest(), + } + config_path = mzml_path.with_name(mzml_path.stem + ".smiter_params.json") + config_path.write_text(json.dumps(config, sort_keys=True, indent=2, default=str)) + sidecar = sign_mzml_output( + mzml_path=mzml_path, + config_path=config_path, + experiment_name=mzml_path.stem, + tool_name="SMITER", + tool_version=smiter.__version__, + ) + logger.info(f"Signed {mzml_path} with mzprov: {sidecar}") + return sidecar + + # @profile def rescale_intensity( i: float, rt: float, molecule: str, peak_properties: dict, isotopologue_lib: dict diff --git a/tests/test_synthetic_mzml.py b/tests/test_synthetic_mzml.py index 3166d11..3da58cc 100644 --- a/tests/test_synthetic_mzml.py +++ b/tests/test_synthetic_mzml.py @@ -1,3 +1,5 @@ +import hashlib +import json from tempfile import NamedTemporaryFile import numpy as np @@ -619,3 +621,59 @@ def test_dynacmic_exclusion(): assert number_fragment_specs == 1 # breakpoint() + + +def _inosine_mzml(path, **kwargs): + peak_props = { + "inosine": { + "chemical_formula": "+C(10)H(12)N(4)O(5)", + "trivial_name": "inosine", + "charge": 2, + "scan_start_time": 0, + "peak_width": 30, # seconds + "peak_function": None, + "peak_params": {}, + } + } + return write_mzml( + str(path), + peak_props, + fragmentor, + GaussNoiseInjector(variance=0.0), + {"gradient_length": 30}, + **kwargs, + ) + + +def test_write_mzml_signed_with_mzprov(tmp_path, monkeypatch): + """sign_with_mzprov writes a sidecar that verifies, and binds the ground truth.""" + mzprov = pytest.importorskip("mzprov") + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config")) + + mzml_path = _inosine_mzml(tmp_path / "inosine.mzML", sign_with_mzprov=True) + + sidecar = tmp_path / "inosine.provenance.json" + assert (tmp_path / "inosine.smiter_params.json").is_file() + result = mzprov.verify_sidecar(sidecar) + assert result.overall_ok + assert result.payload.tool_name == "SMITER" + assert result.payload.tool_version == smiter.__version__ + assert pymzml.run.Reader(mzml_path).get_spectrum_count() == 1000 + + # The signed parameters record the ground truth's hash. + params = json.loads((tmp_path / "inosine.smiter_params.json").read_text()) + summary = (tmp_path / "molecule_summary.csv").read_bytes() + assert params["molecule_summary_sha256"] == hashlib.sha256(summary).hexdigest() + + # Editing the signed copy of the parameters afterwards fails verification. + signed_copy = tmp_path / "inosine.config.toml" + assert json.loads(signed_copy.read_text()) == params + params["mzml_params"]["gradient_length"] = 60 + signed_copy.write_text(json.dumps(params)) + assert not mzprov.verify_sidecar(sidecar).overall_ok + + +def test_write_mzml_unsigned_by_default(tmp_path): + _inosine_mzml(tmp_path / "plain.mzML") + assert not (tmp_path / "plain.provenance.json").exists() + assert not (tmp_path / "plain.smiter_params.json").exists()