Add Wikidata compound identifier conversions

MSMetaEnhancer/libs/converters/web/Wikidata.py

@@ -0,0 +1,179 @@
+import json
+from urllib.parse import urlencode
+
+from frozendict import frozendict
+
+from MSMetaEnhancer.libs.converters.web.WebConverter import WebConverter
+
+
+class Wikidata(WebConverter):
+    """
+    Convert compound names, identifiers, and structures using Wikidata.
+
+    Wikidata Query Service: https://query.wikidata.org/
+    """
+
+    attributes = (
+        "compound_name",
+        "casno",
+        "inchi",
+        "inchikey",
+        "canonical_smiles",
+        "pubchemid",
+        "wikidataid",
+    )
+    identifier_properties = {
+        "pubchemid": "P662",
+        "inchi": "P234",
+        "inchikey": "P235",
+    }
+
+    def __init__(self, session):
+        super().__init__(session)
+        self.endpoints = {"Wikidata": "https://query.wikidata.org/sparql?"}
+        source_methods = {
+            "compound_name": "from_name",
+            "casno": "from_casno",
+            "wikidataid": "from_wikidataid",
+            "pubchemid": "from_pubchemid",
+            "inchi": "from_inchi",
+            "inchikey": "from_inchikey",
+        }
+        conversions = [
+            (source, target, method)
+            for source, method in source_methods.items()
+            for target in self.attributes
+            if source != target
+        ]
+        self.create_top_level_conversion_methods(conversions)
+
+    @staticmethod
+    def result_variables():
+        return " ".join(f"?{attribute}" for attribute in Wikidata.attributes)
+
+    @staticmethod
+    def item_attributes_query():
+        return """
+              OPTIONAL {
+                ?item rdfs:label ?compound_name .
+                FILTER(LANG(?compound_name) = "en")
+              }
+              OPTIONAL { ?item wdt:P231 ?casno . }
+              OPTIONAL { ?item wdt:P234 ?inchi . }
+              OPTIONAL { ?item wdt:P235 ?inchikey . }
+              OPTIONAL { ?item wdt:P233 ?canonical_smiles . }
+              OPTIONAL { ?item wdt:P662 ?pubchemid . }
+              BIND(STRAFTER(STR(?item), "http://www.wikidata.org/entity/")
+                   AS ?wikidataid)
+        """
+
+    async def from_name(self, name):
+        query = f"""
+            PREFIX bd: <http://www.bigdata.com/rdf#>
+            PREFIX mwapi: <https://www.mediawiki.org/ontology#API/>
+            PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
+            PREFIX wikibase: <http://wikiba.se/ontology#>
+            PREFIX wdt: <http://www.wikidata.org/prop/direct/>
+
+            SELECT {self.result_variables()} WHERE {{
+              SERVICE wikibase:mwapi {{
+                bd:serviceParam wikibase:api "EntitySearch" ;
+                  wikibase:endpoint "www.wikidata.org" ;
+                  mwapi:search {self.sparql_literal(name)} ;
+                  mwapi:language "en" ;
+                  mwapi:limit "1" .
+                ?item wikibase:apiOutputItem mwapi:item .
+              }}
+              {self.item_attributes_query()}
+            }}
+            LIMIT 1
+        """
+        return await self.call_service(query)
+
+    async def from_casno(self, casno):
+        query = f"""
+            PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
+            PREFIX wdt: <http://www.wikidata.org/prop/direct/>
+
+            SELECT {self.result_variables()} WHERE {{
+              ?item wdt:P231 ?casno .
+              FILTER(?casno = {self.sparql_literal(casno)})
+              {self.item_attributes_query()}
+            }}
+            LIMIT 1
+        """
+        return await self.call_service(query)
+
+    async def from_wikidataid(self, wikidataid):
+        if not wikidataid.startswith("Q") or not wikidataid[1:].isdigit():
+            return {}
+
+        query = f"""
+            PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
+            PREFIX wd: <http://www.wikidata.org/entity/>
+            PREFIX wdt: <http://www.wikidata.org/prop/direct/>
+
+            SELECT {self.result_variables()} WHERE {{
+              BIND(wd:{wikidataid} AS ?item)
+              {self.item_attributes_query()}
+            }}
+            LIMIT 1
+        """
+        return await self.call_service(query)
+
+    async def from_pubchemid(self, pubchemid):
+        return await self.from_identifier("pubchemid", pubchemid)
+
+    async def from_inchi(self, inchi):
+        return await self.from_identifier("inchi", inchi)
+
+    async def from_inchikey(self, inchikey):
+        return await self.from_identifier("inchikey", inchikey)
+
+    async def from_identifier(self, attribute, value):
+        property_id = self.identifier_properties[attribute]
+        query = f"""
+            PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
+            PREFIX wdt: <http://www.wikidata.org/prop/direct/>
+
+            SELECT {self.result_variables()} WHERE {{
+              ?item wdt:{property_id} ?identifier .
+              FILTER(STR(?identifier) = {self.sparql_literal(value)})
+              {self.item_attributes_query()}
+            }}
+            LIMIT 1
+        """
+        return await self.call_service(query)
+
+    async def call_service(self, query):
+        args = urlencode({"query": query, "format": "json"})
+        response = await self.query_the_service(
+            "Wikidata",
+            args,
+            headers=frozendict(
+                {
+                    "User-Agent": (
+                        "MSMetaEnhancer (https://github.com/RECETOX/MSMetaEnhancer)"
+                    )
+                }
+            ),
+        )
+        if response:
+            return self.parse_attributes(response)
+        return {}
+
+    @staticmethod
+    def sparql_literal(value):
+        return json.dumps(str(value), ensure_ascii=False)
+
+    @staticmethod
+    def parse_attributes(response):
+        bindings = json.loads(response).get("results", {}).get("bindings", [])
+        if not bindings:
+            return {}
+
+        return {
+            attribute: bindings[0][attribute]["value"]
+            for attribute in Wikidata.attributes
+            if attribute in bindings[0]
+        }

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.Wikidata import Wikidata
 
-__all__ = ["IDSM", "CTS", "CIR", "PubChem", "BridgeDb"]
+__all__ = ["IDSM", "CTS", "CIR", "PubChem", "BridgeDb", "Wikidata"]

README.md

@@ -7,7 +7,7 @@
 [![DOI](https://joss.theoj.org/papers/10.21105/joss.04494/status.svg)](https://doi.org/10.21105/joss.04494)
 
 **MSMetaEnhancer** is a tool used for `.msp` files annotation.
-It adds metadata like SMILES, InChI, and CAS number fetched from the following services: [CIR](https://cactus.nci.nih.gov/chemical/structure_documentation), [CTS](https://cts.fiehnlab.ucdavis.edu/), [PubChem](https://pubchem.ncbi.nlm.nih.gov/), [IDSM](https://idsm.elixir-czech.cz/), and [BridgeDb](https://bridgedb.github.io/).
+It adds metadata like SMILES, InChI, and CAS number fetched from the following services: [CIR](https://cactus.nci.nih.gov/chemical/structure_documentation), [CTS](https://cts.fiehnlab.ucdavis.edu/), [PubChem](https://pubchem.ncbi.nlm.nih.gov/), [IDSM](https://idsm.elixir-czech.cz/), [BridgeDb](https://bridgedb.github.io/), and [Wikidata](https://www.wikidata.org/).
 The app uses asynchronous implementation of annotation process allowing for optimal fetching speed.
 
 If you use MSMetaEnhancer in your work, please cite the following publication:
@@ -20,11 +20,18 @@ Troják et al., (2022). MSMetaEnhancer: A Python package for mass spectra metada
 import asyncio
 
 from MSMetaEnhancer import Application
-from MSMetaEnhancer.libs.converters.web import CTS, CIR, IDSM, PubChem, BridgeDb
+from MSMetaEnhancer.libs.converters.web import (
+    CTS,
+    CIR,
+    IDSM,
+    PubChem,
+    BridgeDb,
+    Wikidata,
+)
 from MSMetaEnhancer.libs.converters.compute import RDKit
 from MSMetaEnhancer.libs.utils.ConverterBuilder import ConverterBuilder
 
-ConverterBuilder.register([CTS, CIR, IDSM, PubChem, BridgeDb, RDKit])
+ConverterBuilder.register([CTS, CIR, IDSM, PubChem, BridgeDb, Wikidata, RDKit])
 
 app = Application()
 
@@ -35,7 +42,7 @@ app.load_data('sample.msp', file_format='msp')
 app.curate_metadata()
 
 # specify requested services (these are supported)
-services = ['CTS', 'CIR', 'IDSM', 'PubChem', 'BridgeDb', 'RDKit']
+services = ['CTS', 'CIR', 'IDSM', 'PubChem', 'BridgeDb', 'Wikidata', 'RDKit']
 
 # specify requested jobs
 jobs = [('name', 'inchi', 'IDSM'), ('inchi', 'formula', 'IDSM'), ('inchi', 'inchikey', 'IDSM'),

docs/source/MSMetaEnhancer.libs.webConverters.rst

@@ -41,6 +41,14 @@ BridgeDb
    :undoc-members:
    :show-inheritance:
 
+Wikidata
+--------
+
+.. automodule:: MSMetaEnhancer.libs.converters.web.Wikidata
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
 General Web Converter
 ---------------------
 

galaxy/generate_options.py

@@ -4,7 +4,14 @@
 # this add to path the home dir, so it can be called from anywhere
 sys.path.append(os.path.split(sys.path[0])[0])
 
-from MSMetaEnhancer.libs.converters.web import CTS, CIR, IDSM, PubChem, BridgeDb
+from MSMetaEnhancer.libs.converters.web import (
+    CTS,
+    CIR,
+    IDSM,
+    PubChem,
+    BridgeDb,
+    Wikidata,
+)
 from MSMetaEnhancer.libs.converters.compute import RDKit
 from MSMetaEnhancer.libs.utils.ConverterBuilder import ConverterBuilder
 
@@ -13,7 +20,7 @@
 
 
 def generate_options():
-    ConverterBuilder.register([CTS, CIR, IDSM, PubChem, BridgeDb, RDKit])
+    ConverterBuilder.register([CTS, CIR, IDSM, PubChem, BridgeDb, Wikidata, RDKit])
 
     jobs = []
     converters = web_converters + compute_converters