diff --git a/MSMetaEnhancer/libs/converters/web/ChEBI.py b/MSMetaEnhancer/libs/converters/web/ChEBI.py new file mode 100644 index 0000000..5bc6a02 --- /dev/null +++ b/MSMetaEnhancer/libs/converters/web/ChEBI.py @@ -0,0 +1,276 @@ +import json +from urllib.parse import quote +from frozendict import frozendict + +from MSMetaEnhancer.libs.converters.web.WebConverter import WebConverter + + +class ChEBI(WebConverter): + """ + ChEBI (Chemical Entities of Biological Interest) is a freely available dictionary of + molecular entities focused on small chemical compounds. + + ChEBI service: https://www.ebi.ac.uk/chebi/ + API documentation: https://www.ebi.ac.uk/chebi/backend/api/docs/ + """ + + def __init__(self, session): + super().__init__(session) + # service URLs + self.endpoints = { + "ChEBI": "https://www.ebi.ac.uk/chebi/backend/api/public/", + } + + self.attributes = [ + {"code": "chebiid", "paths": [("chebi_accession",), ("chebiId",)]}, + { + "code": "compound_name", + "paths": [("ascii_name",), ("name",), ("chebiAsciiName",)], + }, + { + "code": "iupac_name", + "paths": [("iupac_names",), ("iupacNames",), ("iupacName",), ("iupac_name",)], + }, + { + "code": "inchikey", + "paths": [ + ("default_structure", "standard_inchi_key"), + ("inchikey",), + ("standard_inchi_key",), + ("inchiKey",), + ], + }, + { + "code": "inchi", + "paths": [("default_structure", "standard_inchi"), ("inchi",)], + }, + {"code": "smiles", "paths": [("default_structure", "smiles"), ("smiles",)]}, + { + "code": "formula", + "paths": [("chemical_data", "formula"), ("formula",)], + }, + ] + + # generate top level methods defining allowed conversions + conversions = [ + ("compound_name", "chebiid", "from_name"), + ("compound_name", "inchikey", "from_name"), + ("compound_name", "inchi", "from_name"), + ("compound_name", "smiles", "from_name"), + ("compound_name", "formula", "from_name"), + ("iupac_name", "chebiid", "from_iupac_name"), + ("iupac_name", "compound_name", "from_iupac_name"), + ("iupac_name", "inchikey", "from_iupac_name"), + ("iupac_name", "inchi", "from_iupac_name"), + ("iupac_name", "smiles", "from_iupac_name"), + ("iupac_name", "formula", "from_iupac_name"), + ("inchikey", "chebiid", "from_inchikey"), + ("inchikey", "compound_name", "from_inchikey"), + ("inchikey", "iupac_name", "from_inchikey"), + ("inchikey", "inchi", "from_inchikey"), + ("inchikey", "smiles", "from_inchikey"), + ("inchikey", "formula", "from_inchikey"), + ("inchi", "chebiid", "from_inchi"), + ("inchi", "compound_name", "from_inchi"), + ("inchi", "iupac_name", "from_inchi"), + ("inchi", "inchikey", "from_inchi"), + ("inchi", "smiles", "from_inchi"), + ("inchi", "formula", "from_inchi"), + ("smiles", "chebiid", "from_smiles"), + ("smiles", "compound_name", "from_smiles"), + ("smiles", "iupac_name", "from_smiles"), + ("smiles", "inchikey", "from_smiles"), + ("smiles", "inchi", "from_smiles"), + ("smiles", "formula", "from_smiles"), + ("chebiid", "compound_name", "from_chebiid"), + ("chebiid", "iupac_name", "from_chebiid"), + ("chebiid", "inchikey", "from_chebiid"), + ("chebiid", "inchi", "from_chebiid"), + ("chebiid", "smiles", "from_chebiid"), + ("chebiid", "formula", "from_chebiid"), + ] + self.create_top_level_conversion_methods(conversions) + + async def from_name(self, name): + """ + Convert compound name to all possible attributes using ChEBI service. + + :param name: given compound name + :return: all found data + """ + return await self._from_es_search(name) + + async def from_iupac_name(self, iupac_name): + """ + Convert IUPAC name to all possible attributes using ChEBI service. + + :param iupac_name: given IUPAC name + :return: all found data + """ + return await self._from_es_search(iupac_name) + + async def _from_es_search(self, term): + """ + Search ChEBI by a given term and return parsed attributes. + + :param term: search term (name, InChIKey, InChI, SMILES, or IUPAC name) + :return: all found data + """ + args = f"es_search/?term={quote(term)}&page=1&size=15" + response = await self.query_the_service("ChEBI", args) + if response: + return self.parse_search_results(response) + + async def from_inchikey(self, inchikey): + """ + Convert InChIKey to all possible attributes using ChEBI service. + + :param inchikey: given InChIKey + :return: all found data + """ + return await self._from_es_search(inchikey) + + async def from_inchi(self, inchi): + """ + Convert InChI to all possible attributes using ChEBI service. + + :param inchi: given InChI string + :return: all found data + """ + return await self._from_es_search(inchi) + + async def from_smiles(self, smiles): + """ + Convert SMILES to all possible attributes using ChEBI structure search service. + + Uses the similarity search endpoint with a similarity of 1.0 (exact match). + + :param smiles: given SMILES string + :return: all found data + """ + data = frozendict({"structure": smiles, "type": "similarity", "similarity": 1.0}) + response = await self.query_the_service( + "ChEBI", "structure_search/", method="POST", data=data + ) + if response: + return self.parse_structure_search_results(response) + + async def from_chebiid(self, chebiid): + """ + Convert ChEBI ID to all possible attributes using ChEBI service. + + :param chebiid: given ChEBI ID (e.g. 'CHEBI:60888') + :return: all found data + """ + args = f"compound/{chebiid}/" + response = await self.query_the_service("ChEBI", args) + if response: + return self.parse_entity(response) + + def parse_entity(self, response): + """ + Parse attributes from a single ChEBI entity response. + + :param response: JSON string from /compound/{chebiId}/ endpoint + :return: dict of parsed attributes + """ + entity = json.loads(response) + return self._extract_attributes(entity) + + def parse_structure_search_results(self, response): + """ + Parse attributes from the first result of a ChEBI structure search response. + + The structure_search endpoint returns a list of compound objects. + + :param response: JSON string from /structure_search/ endpoint + :return: dict of parsed attributes from the first result + """ + results = json.loads(response) + if not isinstance(results, list) or not results: + return None + return self._extract_attributes(results[0]) + + def parse_search_results(self, response): + """ + Parse attributes from the first result of a ChEBI search response. + + :param response: JSON string from /es_search/ endpoint + :return: dict of parsed attributes from the first result + """ + response_json = json.loads(response) + results = response_json.get("results", []) + if not results: + return None + entity = results[0] + if isinstance(entity, dict): + entity = entity.get("_source", entity) + return self._extract_attributes(entity) + + def _extract_attributes(self, entity): + """ + Extract known attributes from a ChEBI entity dict. + + :param entity: dict representing a ChEBI entity + :return: dict of parsed attributes + """ + if not isinstance(entity, dict): + return None + result = {} + for att in self.attributes: + value = self._get_first_value(entity, att["paths"]) + if value is not None: + result[att["code"]] = value + # Extract IUPAC name from names dict (entity endpoint) or synonyms list (legacy) + if "iupac_name" not in result: + iupac_name = self._extract_iupac_from_names(entity) + if iupac_name is not None: + result["iupac_name"] = iupac_name + return result if result else None + + def _extract_iupac_from_names(self, entity): + """ + Extract IUPAC name from the entity response. + + Tries two structures: + - ``names["IUPAC NAME"][0]["name"]`` (entity endpoint format) + - ``synonyms[].type == "IUPAC NAME"`` → ``.data`` (legacy/es_search format) + + :param entity: dict representing a ChEBI entity + :return: IUPAC name string or None + """ + names = entity.get("names", {}) + if isinstance(names, dict): + iupac_entries = names.get("IUPAC NAME", []) + if isinstance(iupac_entries, list) and iupac_entries: + entry = iupac_entries[0] + if isinstance(entry, dict): + return entry.get("name") or entry.get("ascii_name") + synonyms = entity.get("synonyms", []) + if isinstance(synonyms, list): + for syn in synonyms: + if isinstance(syn, dict) and syn.get("type", "").upper() == "IUPAC NAME": + return syn.get("data") + return None + + def _get_first_value(self, entity, paths): + """ + Return the first non-empty value found in the given candidate paths. + If the resolved value is a list, the first element is returned. + + :param entity: dict representing a ChEBI entity + :param paths: candidate key paths to try + :return: first non-empty value or None + """ + for path in paths: + value = entity + for key in path: + if not isinstance(value, dict): + value = None + break + value = value.get(key) + if value is not None: + if isinstance(value, list): + return next((v for v in value if v), None) + return value + return None diff --git a/MSMetaEnhancer/libs/converters/web/__init__.py b/MSMetaEnhancer/libs/converters/web/__init__.py index 76e1e31..61fb420 100644 --- a/MSMetaEnhancer/libs/converters/web/__init__.py +++ b/MSMetaEnhancer/libs/converters/web/__init__.py @@ -3,5 +3,6 @@ from MSMetaEnhancer.libs.converters.web.CIR import CIR from MSMetaEnhancer.libs.converters.web.PubChem import PubChem from MSMetaEnhancer.libs.converters.web.BridgeDb import BridgeDb +from MSMetaEnhancer.libs.converters.web.ChEBI import ChEBI -__all__ = ["IDSM", "CTS", "CIR", "PubChem", "BridgeDb"] +__all__ = ["IDSM", "CTS", "CIR", "PubChem", "BridgeDb", "ChEBI"] diff --git a/tests/test_ChEBI.py b/tests/test_ChEBI.py new file mode 100644 index 0000000..5b373b4 --- /dev/null +++ b/tests/test_ChEBI.py @@ -0,0 +1,252 @@ +import asyncio +import json +import pytest + +from MSMetaEnhancer.libs.converters.web import ChEBI +from tests.utils import wrap_with_session + + +CHEBI_ID = "CHEBI:60888" +INCHIKEY = "FTEDXVNDVHYDQW-UHFFFAOYSA-N" +INCHI = "InChI=1S/C22H24N2O10/c25-19(26)11-23(12-20(27)28)15-5-1-3-7-17(15)33-9-10-34-18-8-4-2-6-16(18)24(13-21(29)30)14-22(31)32/h1-8H,9-14H2,(H,25,26)(H,27,28)(H,29,30)(H,31,32)" +COMPOUND_NAME = "bapta" +IUPAC_NAME = "2,2',2'',2'''-[ethane-1,2-diylbis(oxy-2,1-phenylenenitrilo)]tetraacetic acid" +SMILES = "OC(=O)CN(CCOc1ccccc1N(CC(O)=O)CC(O)=O)CC(O)=O" + + +@pytest.mark.dependency() +def test_service_available(): + asyncio.run(wrap_with_session(ChEBI, "chebiid_to_inchikey", [CHEBI_ID])) + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_chebiid_to_inchikey(): + result = asyncio.run(wrap_with_session(ChEBI, "chebiid_to_inchikey", [CHEBI_ID])) + assert result is not None + assert "inchikey" in result + assert result["inchikey"] == INCHIKEY + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_chebiid_to_smiles(): + result = asyncio.run(wrap_with_session(ChEBI, "chebiid_to_smiles", [CHEBI_ID])) + assert result is not None + assert "smiles" in result + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_smiles_to_chebiid(): + result = asyncio.run(wrap_with_session(ChEBI, "smiles_to_chebiid", [SMILES])) + assert result is not None + assert "chebiid" in result + assert result["chebiid"] == CHEBI_ID + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_chebiid_to_inchi(): + result = asyncio.run(wrap_with_session(ChEBI, "chebiid_to_inchi", [CHEBI_ID])) + assert result is not None + assert "inchi" in result + assert result["inchi"] == INCHI + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_chebiid_to_iupac_name(): + result = asyncio.run(wrap_with_session(ChEBI, "chebiid_to_iupac_name", [CHEBI_ID])) + assert result is not None + assert "iupac_name" in result + assert result["iupac_name"] == IUPAC_NAME + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_inchi_to_chebiid(): + result = asyncio.run(wrap_with_session(ChEBI, "inchi_to_chebiid", [INCHI])) + assert result is not None + assert "chebiid" in result + assert result["chebiid"] == CHEBI_ID + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_iupac_name_to_chebiid(): + result = asyncio.run( + wrap_with_session(ChEBI, "iupac_name_to_chebiid", [IUPAC_NAME]) + ) + assert result is not None + assert "chebiid" in result + assert result["chebiid"] == CHEBI_ID + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_chebiid_to_formula(): + result = asyncio.run(wrap_with_session(ChEBI, "chebiid_to_formula", [CHEBI_ID])) + assert result is not None + assert "formula" in result + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_inchikey_to_chebiid(): + result = asyncio.run(wrap_with_session(ChEBI, "inchikey_to_chebiid", [INCHIKEY])) + assert result is not None + assert "chebiid" in result + assert result["chebiid"] == CHEBI_ID + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_compound_name_to_chebiid(): + result = asyncio.run( + wrap_with_session(ChEBI, "compound_name_to_chebiid", [COMPOUND_NAME]) + ) + assert result is not None + assert "chebiid" in result + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_format_search_response(): + args = f"es_search/?term={INCHIKEY}&page=1&size=15" + response = asyncio.run( + wrap_with_session(ChEBI, "query_the_service", ["ChEBI", args]) + ) + + assert isinstance(response, str) + data = json.loads(response) + assert "results" in data + + +@pytest.mark.dependency(depends=["test_service_available"]) +def test_format_entity_response(): + args = f"compound/{CHEBI_ID}/" + response = asyncio.run( + wrap_with_session(ChEBI, "query_the_service", ["ChEBI", args]) + ) + + assert isinstance(response, str) + data = json.loads(response) + assert "chebi_accession" in data + assert data["chebi_accession"] == CHEBI_ID + + +def test_parse_entity_nested_response(): + response = json.dumps( + { + "chebi_accession": CHEBI_ID, + "name": "bapta", + "chemical_data": {"formula": "C22H22N2O10"}, + "default_structure": { + "smiles": "C1=CC=C(C=C1)O", + "standard_inchi": "InChI=1S/C6H6O/c7-6-4-2-1-3-5-6/h1-5,7H", + "standard_inchi_key": INCHIKEY, + }, + "names": { + "IUPAC NAME": [ + { + "name": IUPAC_NAME, + "type": "IUPAC NAME", + "source": "IUPAC", + "ascii_name": IUPAC_NAME, + } + ], + "SYNONYM": [ + {"name": "bapta", "type": "SYNONYM"}, + ], + }, + } + ) + + parsed = ChEBI(None).parse_entity(response) + + assert parsed == { + "chebiid": CHEBI_ID, + "compound_name": "bapta", + "iupac_name": IUPAC_NAME, + "formula": "C22H22N2O10", + "smiles": "C1=CC=C(C=C1)O", + "inchi": "InChI=1S/C6H6O/c7-6-4-2-1-3-5-6/h1-5,7H", + "inchikey": INCHIKEY, + } + + +def test_parse_entity_synonyms_fallback(): + """Ensure iupac_name can also be extracted from legacy synonyms list format.""" + response = json.dumps( + { + "chebi_accession": CHEBI_ID, + "name": "bapta", + "synonyms": [ + {"type": "IUPAC NAME", "data": IUPAC_NAME}, + {"type": "Synonym", "data": "bapta"}, + ], + } + ) + + parsed = ChEBI(None).parse_entity(response) + + assert parsed is not None + assert parsed["iupac_name"] == IUPAC_NAME + + +def test_parse_search_response_source(): + response = json.dumps( + { + "results": [ + { + "_source": { + "chebi_accession": CHEBI_ID, + "ascii_name": COMPOUND_NAME, + "formula": "C22H22N2O10", + "smiles": "NCCO", + "inchi": "InChI=1S/C2H7NO/c3-1-2-4/h4H,1-3H2", + "standard_inchi_key": INCHIKEY, + "iupac_names": [IUPAC_NAME], + } + } + ] + } + ) + + parsed = ChEBI(None).parse_search_results(response) + + assert parsed == { + "chebiid": CHEBI_ID, + "compound_name": COMPOUND_NAME, + "iupac_name": IUPAC_NAME, + "formula": "C22H22N2O10", + "smiles": "NCCO", + "inchi": "InChI=1S/C2H7NO/c3-1-2-4/h4H,1-3H2", + "inchikey": INCHIKEY, + } + + +def test_get_conversions(): + jobs = ChEBI(None).get_conversion_functions() + assert ("chebiid", "inchikey", "ChEBI") in jobs + assert ("inchikey", "chebiid", "ChEBI") in jobs + assert ("compound_name", "chebiid", "ChEBI") in jobs + assert ("chebiid", "iupac_name", "ChEBI") in jobs + assert ("iupac_name", "chebiid", "ChEBI") in jobs + assert ("inchi", "chebiid", "ChEBI") in jobs + assert ("smiles", "chebiid", "ChEBI") in jobs + + +def test_parse_structure_search_response(): + response = json.dumps( + [ + { + "chebi_accession": CHEBI_ID, + "ascii_name": COMPOUND_NAME, + "chemical_data": {"formula": "C22H24N2O10"}, + "default_structure": { + "smiles": SMILES, + "standard_inchi": INCHI, + "standard_inchi_key": INCHIKEY, + }, + } + ] + ) + + parsed = ChEBI(None).parse_structure_search_results(response) + + assert parsed is not None + assert parsed["chebiid"] == CHEBI_ID + assert parsed["compound_name"] == COMPOUND_NAME + assert parsed["smiles"] == SMILES + assert parsed["inchikey"] == INCHIKEY