diff --git a/README.md b/README.md index e10808e..6f037dd 100644 --- a/README.md +++ b/README.md @@ -48,8 +48,8 @@ For less loading time and memory footprint, a part of the cache can be loaded: ```python from rr_cache import rrCache -cache = rrCache(['cid_strc']) -cache.get_compound('MNXM2') +cache = rrCache(["cid_strc"]) +cache.get_compound("MNXM2") ``` ***From CLI*** ```sh diff --git a/environment.yaml b/environment.yaml index 23846ae..f0d1b35 100644 --- a/environment.yaml +++ b/environment.yaml @@ -7,3 +7,4 @@ dependencies: - requests - rdkit - colored + - pandas diff --git a/rr_cache/__main__.py b/rr_cache/__main__.py index fd2d7cf..1bd9fc4 100644 --- a/rr_cache/__main__.py +++ b/rr_cache/__main__.py @@ -1,16 +1,18 @@ +import sys +from json import dumps +from logging import Logger, getLogger + +from brs_utils import build_args_parser +from brs_utils import init as init_logger +from colored import attr, fg + +from rr_cache.args import add_arguments from rr_cache.rr_cache import ( rrCache, ) -from rr_cache.Args import add_arguments -from brs_utils import init as init_logger, build_args_parser -from logging import Logger, getLogger -from colored import fg, attr -from json import dumps -from typing import ( - List, -) -from .Args import CONFIG_PATH + from ._version import __version__ +from .args import CONFIG_PATH def disable_rdkit_logging(): @@ -55,7 +57,7 @@ def entry_point(): for dt in cspaces: print(f"- {dt}") print() - exit(0) + sys.exit(0) cache = rrCache( cspace=args.cspace, @@ -88,12 +90,15 @@ def entry_point(): ) +LOGGER = getLogger(__name__) + + def print_attr( cache: "rrCache", attr: str, - attr_lst: List, + attr_lst: list, do_not_dwnl_cache: bool, - logger: Logger = getLogger(__file__), + logger: Logger = LOGGER, ) -> None: cache.Load(attrs=[attr], do_not_dwnl_cache=do_not_dwnl_cache) if attr_lst == []: @@ -103,9 +108,7 @@ def print_attr( try: print(id + ":", dumps(cache.get(attr)[id], indent=4)) except KeyError: - logger.error( - "ID not found in rrCache('{attr}'): {id}".format(attr=attr, id=id) - ) + logger.error(f"ID not found in rrCache('{attr}'): {id}") if __name__ == "__main__": diff --git a/rr_cache/Args.py b/rr_cache/args.py similarity index 96% rename from rr_cache/Args.py rename to rr_cache/args.py index 324407f..516fc60 100644 --- a/rr_cache/Args.py +++ b/rr_cache/args.py @@ -5,7 +5,7 @@ CONFIG_PATH = os_path.join(HERE, "config") # Default values for the arguments DEFAULTS = { - "cspace": "rr2026", + "cspace": "rr2026-v3.1.0", "databases": ["metanetx", "rhea"], "interactive": False, "do_not_dwnl_cache": False, @@ -21,7 +21,7 @@ def add_arguments(parser: ArgumentParser) -> ArgumentParser: dest="cspace", default=DEFAULTS["cspace"], type=str, - help="chemical space to use (e.g. mnx3.1, mnx4.0...). Determines which configuration files and folders to use both the cache and the input cache (default: %(default)s).", + help="chemical space to use (e.g. mnx3.1, mnx4.4...). Determines which configuration files and folders to use both the cache and the input cache (default: %(default)s).", ) parser.add_argument( "--list-chemical-spaces", diff --git a/rr_cache/config/config_rr2026.json b/rr_cache/config/config_rr2026-v3.0.json similarity index 100% rename from rr_cache/config/config_rr2026.json rename to rr_cache/config/config_rr2026-v3.0.json diff --git a/rr_cache/config/config_rr2026-v3.1.0.json b/rr_cache/config/config_rr2026-v3.1.0.json new file mode 100644 index 0000000..94b60a9 --- /dev/null +++ b/rr_cache/config/config_rr2026-v3.1.0.json @@ -0,0 +1,77 @@ +{ + "type": "rr2026", + "cache": { + "cid_strc": { + "deps": { + "file_deps": ["chemicals_metadata", "chem_prop.tsv"] + }, + "file": { + "url": "", + "name": "cid_strc.json.gz" + } + }, + "cid_xref": { + "deps": { + "attr_deps": [], + "file_deps": ["chem_xref.tsv"] + }, + "file": { + "url": "https://gitlab.com/breakthewall/rrCache-data/-/raw/master/cache/mnx_4.4/", + "name": "cid_xref.json.gz" + } + }, + "rr_reactions": { + "deps": { + "file_deps": ["templates_metadata"] + }, + "file": { + "url": "", + "name": "rr_reactions.json.gz" + } + }, + "template_reactions": { + "deps": { + "file_deps": ["templates_metadata", "reac_prop.tsv"] + }, + "file": { + "url": "", + "name": "template_reactions.json.gz" + } + } + }, + "sources": { + "mnx": { + "url": "https://www.metanetx.org/ftp/4.5/", + "files": { + "chem_xref.tsv": "2b88dfe3b990ee85c8d8eb51f6d7079bf042b14e4cd11a4cc6ceb841686f9ea7ef4a04358363561fbbb2187a9870a1930037a030283908788a572e7b96b91ec7", + "chem_prop.tsv": "00d8d09d52884961748d032288781c4d47929f12d5052b367116c411194fdfae879b15844e8a15bccd494c26702dae57fd682c3bc03b5fee3c3e92e138c1eac2", + "reac_prop.tsv": "1796b5d77c707d0cfbfce99cf0466ef6aff0fe61ea5ded043216aa334e5ecd1388364d214162342f470b51eca2dab4309dcec2b9e074abee47dcae974c674bd7" + } + }, + "rr2": { + "url": "https://retrorules.org/dl/v3.1.0/", + "files": { + "templates_metadata": { + "metanetx": "25cf9e9e4162cb9039e8692787890568c2d45e82cd2c6f28ed574a6643fc48e0b61e196b9e552419cd68dc5c86a8c2aa08c1f6731b2543e35387ebbde5587867", + "rhea": "37fb906124fdb61a12b378a861341c64ecea459e8f3e8c3a5f0865290da0b712bb6b07aada4bf358ca2bdb65fd3faa5cd6acd9380ae26c64e14d6cccf70b6085", + "uspto": "6aebfd43a3932b11009630a7e5fdc3e10fde9a20d9933f7792d0904c945ea1624c94a0e24a9d56d0d103b9ecfdddd1e2d8f9edb177886ecac23d5775006bc23e" + } + } + }, + "rr2more": { + "url": "https://retrorules.org/dl/v3.0/", + "files": { + "reactions_metadata": { + "metanetx": "5a61ca70e477f23a9300c3bdfb912ebd0df3a0d985a2d4b46b5a002db812ae468b9b736a72f045d022a577eeda03b81e74f6c044df5fa135dfdfc34b7e26d380", + "rhea": "718295f7804b5d343c6b761550c408b9c3a6d5f1537f1f0de5858ceae6248919bd19ccb5cf4600c108e9004d20b0a4bc54f6ae85415d0dd9d964e1a12585dd10", + "uspto": "4f893d0038e205ec06d66f8165000c07e17e0eb099873c252ed654c9c1cad8afa6cfbb236f60983b4d43601a71678eb469a72f53a1582914e87783c8e5ed51b8" + }, + "chemicals_metadata": { + "metanetx": "d0c618364af5961abb7ce99a93d29222faf3371a7270edd4667e197c87f8f2eed13d4be22a9aa03e7c4b160b424250c37c6279fe51cb68b6aae3ae968472d321", + "rhea": "e46cdffb86c5ae46d3338faff720a77ca4651fe37d645b0072e7fbc78cfa19ba550ba6c2e157ceee28d09ff8e0c1fd6f18a59e6c653d1a4bfe958074f39e97c3", + "uspto": "2c4aefd0c0341a2491fd1be4478092c8cfdbaf7784e5cbf982807631c5ad4c129cdbdbe9e1caebf0ae975570a004f2e498a0971c8fded386f965472bee88081b" + } + } + } + } +} \ No newline at end of file diff --git a/rr_cache/rr_cache.py b/rr_cache/rr_cache.py index af2d5a2..a09d559 100644 --- a/rr_cache/rr_cache.py +++ b/rr_cache/rr_cache.py @@ -1,32 +1,40 @@ -from os import path as os_path, makedirs -from rdkit.Chem import ( - MolFromSmiles, - MolFromInchi, - MolToSmiles, - MolToInchi, - MolToInchiKey, -) -from csv import DictReader as csv_DictReader, reader as csv_reader -from pandas import ( - read_csv as pd_read_csv, - DataFrame, -) +import sys +from collections import Counter +from csv import DictReader as csv_DictReader +from csv import reader as csv_reader +from gzip import GzipFile +from gzip import open as gzip_open +from hashlib import sha512 from io import StringIO -from json import dump as json_dump, dumps as json_dumps, load as json_load -from gzip import open as gzip_open, GzipFile +from json import dump as json_dump +from json import dumps as json_dumps +from json import load as json_load +from logging import Logger, StreamHandler, getLogger +from os import makedirs +from os import path as os_path +from pathlib import Path from re import findall as re_findall -from requests import exceptions as r_exceptions -from hashlib import sha512 -from pathlib import Path +from brs_utils import check_sha, download, print_end, print_progress, print_start from colored import ( attr as c_attr, ) -from logging import Logger, getLogger, StreamHandler -from typing import List, Tuple, Dict -from collections import Counter -from brs_utils import print_start, print_progress, print_end, download, check_sha -from .Args import DEFAULTS, CONFIG_PATH +from pandas import ( + DataFrame, +) +from pandas import ( + read_csv as pd_read_csv, +) +from rdkit.Chem import ( + MolFromInchi, + MolFromSmiles, + MolToInchi, + MolToInchiKey, + MolToSmiles, +) +from requests import exceptions as r_exceptions + +from .args import CONFIG_PATH, DEFAULTS def ask_user_input(): @@ -37,7 +45,7 @@ def ask_user_input(): # Check the user's input if user_input.lower() == "s": print("Stopping the program.") - exit() + sys.exit() elif user_input.lower() == "c": print("Continuing without prompting.") return False @@ -54,6 +62,9 @@ class FingerprintError(Exception): pass +LOGGER = getLogger(__name__) + + class rrCache: """Class to generate the cache @@ -67,12 +78,12 @@ class rrCache: def __init__( self, cspace: str = DEFAULTS["cspace"], - databases: List[str] = DEFAULTS["databases"], + databases: list[str] = DEFAULTS["databases"], interactive: bool = DEFAULTS["interactive"], do_not_dwnl_cache: bool = DEFAULTS["do_not_dwnl_cache"], load: bool = True, install_dir: str = DEFAULTS["install_dir"], - logger: Logger = getLogger(__name__), + logger: Logger = LOGGER, ) -> "rrCache": """Constructor for the class Args: @@ -117,7 +128,7 @@ def __init__( f"Cache config file {cache_cfg_fln} not found, please check the --chemical-space argument" ) logger.error("Exiting...") - exit(1) + sys.exit(1) # Cache elements list rrCache.__attributes_list = list(rrCache.__cache.keys()) @@ -160,7 +171,7 @@ def get_type(self) -> str: def Load( self, - attrs: List = [], + attrs: list | None = None, interactive: bool = DEFAULTS["interactive"], do_not_dwnl_cache: bool = DEFAULTS["do_not_dwnl_cache"], type: str = "legacy", @@ -177,7 +188,7 @@ def Load( self.logger.debug("do_not_dwnl_cache: " + str(do_not_dwnl_cache)) if attrs is None: - return + attrs = [] if attrs != []: if not isinstance(attrs, list): @@ -220,7 +231,7 @@ def get_input_cache_dir(self) -> str: @staticmethod def _check_or_download_cache_to_disk( - cache_dir: str, attributes_list: Dict, logger: Logger = getLogger(__name__) + cache_dir: str, attributes_list: dict, logger: Logger = LOGGER ) -> None: logger.debug("cache_dir: " + str(cache_dir)) logger.debug("attributes: " + str(attributes_list)) @@ -281,7 +292,7 @@ def get(self, attr: str): try: mangled = f"_{self.__class__.__name__}__{attr}" return getattr(self, mangled) - except Exception as e: + except AttributeError as e: self.logger.error(str(e)) return None @@ -304,8 +315,8 @@ def get_reaction(self, rxn_id: str): def _m_mnx_reaction_from_reac_prop( reac_prop_path: str, rxn_id: str, - logger: Logger = getLogger(__name__), - ) -> Dict: + logger: Logger = LOGGER, + ) -> dict: if not os_path.exists(reac_prop_path): logger.error(f"MetaNetX reaction file not found: {reac_prop_path}") return None @@ -376,7 +387,7 @@ def __contains__(self, id: str) -> bool: self.Load(attrs=[attr]) if id in self.get(attr): return True - except Exception: + except AttributeError: continue return False @@ -385,7 +396,7 @@ def __get_object(self, attr: str, id: str): if not self.__hasattr(attr): self.Load(attrs=[attr]) return self.get(attr)[id] - except Exception as e: + except AttributeError as e: self.logger.error(str(e)) def __get_list_of_objects(self, attr: str): @@ -393,13 +404,13 @@ def __get_list_of_objects(self, attr: str): if not self.__hasattr(attr): self.Load(attrs=[attr]) return self.get(attr).keys() - except Exception as e: + except AttributeError as e: self.logger.error(str(e)) def set(self, attr: str, val: object): try: return setattr(self, "__" + attr, val) - except Exception as e: + except AttributeError as e: self.logger.error(str(e)) ##################################################### @@ -409,8 +420,6 @@ def set(self, attr: str, val: object): class Error(Exception): """Error function for the convertion of structures""" - pass - class DepictionError(Error): """Error function for the convertion of structures""" @@ -433,7 +442,7 @@ def _download_if_not_exists_or_corrupted( filename: str, outdir: str, fingerprint: str, - logger: Logger = getLogger(__name__), + logger: Logger = LOGGER, ) -> None: # Download if not exists or corrupted if not os_path.exists(os_path.join(outdir, filename)): @@ -496,10 +505,8 @@ def Build(self, interactive: bool = DEFAULTS["interactive"]) -> None: # FETCH INPUT_CACHE FILES print_start(self.logger, "Checking input cache") for input_type, input in rrCache.__cache_sources.items(): - self.logger.debug(f"Checking {input_type}...") for filename, fingerprint in input["files"].items(): - # If fingerprint is a dict, it means that the file is database-specific and the keys are the databases for which the file should be downloaded, otherwise it is a general file to download for all chemical spaces if isinstance(fingerprint, dict): databases = fingerprint.keys() @@ -600,8 +607,8 @@ def Build(self, interactive: bool = DEFAULTS["interactive"]) -> None: @staticmethod def _gen_deprecatedCID_cid( - input_dir: str, outdir: str, logger: Logger = getLogger(__name__) - ) -> Dict: + input_dir: str, outdir: str, logger: Logger = LOGGER + ) -> dict: attribute = "deprecatedCID_cid" logger.debug(c_attr("bold") + attribute + c_attr("reset")) logger.debug(f" input_dir: {input_dir}") @@ -636,9 +643,9 @@ def _gen_cid_strc_cid_name( outdir: str, interactive: bool = DEFAULTS["interactive"], type: str = "legacy", - databases: List[str] = DEFAULTS["databases"], - logger: Logger = getLogger(__name__), - ) -> Dict: + databases: list[str] = DEFAULTS["databases"], + logger: Logger = LOGGER, + ) -> dict: logger.debug(f"input_dir: {input_dir}") logger.debug(f"outdir: {outdir}") @@ -677,24 +684,27 @@ def _gen_cid_strc_cid_name( ) else: dep_files[scat].append(os_path.join(input_dir, dep_file)) - cid_strc, cid_name = rrCache._m_mnxm_strc( + cid_strc, _cid_name = rrCache._m_mnxm_strc( dep_files, interactive=interactive, logger=logger ) logger.debug(" Writing data to file...") rrCache._store_cache_to_file(cid_strc, f_cid_strc, logger=logger) - return {"attr": cid_strc, "file": f_cid_strc}, { - # 'attr': cid_name, - # 'file': f_cid_name - } + return ( + {"attr": cid_strc, "file": f_cid_strc}, + { + # 'attr': cid_name, + # 'file': f_cid_name + }, + ) @staticmethod def _gen_inchikey_cid( input_dir: str, outdir: str, - cid_strc: Dict, - logger: Logger = getLogger(__name__), + cid_strc: dict, + logger: Logger = LOGGER, ) -> None: attribute = "inchikey_cid" logger.debug(c_attr("bold") + attribute + c_attr("reset")) @@ -722,8 +732,8 @@ def _gen_inchikey_cid( def _gen_cid_xref( input_dir: str, outdir: str, - logger: Logger = getLogger(__name__), - ) -> Dict: + logger: Logger = LOGGER, + ) -> dict: logger.debug(f"input_dir: {input_dir}") logger.debug(f"outdir: {outdir}") @@ -756,7 +766,7 @@ def _gen_cid_xref( # input_dir: str, # outdir: str, # cid_xref: Dict, - # logger: Logger = getLogger(__name__), + # logger: Logger = LOGGER, # ) -> Dict: # attribute = "chebi_cid" # logger.debug(c_attr("bold") + attribute + c_attr("reset")) @@ -779,8 +789,8 @@ def _gen_cid_xref( @staticmethod def _gen_deprecatedRID_rid( - input_dir: str, outdir: str, logger: Logger = getLogger(__name__) - ) -> Dict: + input_dir: str, outdir: str, logger: Logger = LOGGER + ) -> dict: attribute = "deprecatedRID_rid" logger.debug(c_attr("bold") + attribute + c_attr("reset")) deprecatedRID_rid = None @@ -811,9 +821,9 @@ def _gen_reactions( input_dir: str, outdir: str, type: str = "legacy", - databases: List[str] = DEFAULTS["databases"], - attribute: str = None, - logger: Logger = getLogger(__name__), + databases: list[str] = DEFAULTS["databases"], + attribute: str | None = None, + logger: Logger = LOGGER, ) -> None: logger.debug(f"input_dir: {input_dir}") logger.debug(f"outdir: {outdir}") @@ -880,7 +890,7 @@ def _gen_reactions( @staticmethod def _gen_comp_xref_deprecatedCompID_compid( - input_dir: str, outdir: str, logger: Logger = getLogger(__name__) + input_dir: str, outdir: str, logger: Logger = LOGGER ) -> None: attribute = "comp_xref, deprecatedCompID_compid" logger.debug(c_attr("bold") + attribute + c_attr("reset")) @@ -938,8 +948,8 @@ def _download_input_cache( url: str, file: str, outdir: str, - fingerprint: str = None, - logger: Logger = getLogger(__name__), + fingerprint: str | None = None, + logger: Logger = LOGGER, ): if not os_path.isdir(outdir): @@ -972,7 +982,7 @@ def _download_input_cache( @staticmethod def __download_input_cache( - url: str, file: str, outdir: str, logger: Logger = getLogger(__name__) + url: str, file: str, outdir: str, logger: Logger = LOGGER ): """Download the input cache file from the given URL to the specified output directory. Args: @@ -1006,13 +1016,14 @@ def __download_input_cache( # @return file content @staticmethod - def _load_json(filename, logger: Logger = getLogger(__name__)): + def _load_json(filename, logger: Logger = LOGGER): logger.debug(filename) - if filename.endswith(".gz") or filename.endswith(".zip"): - fp = gzip_open(filename, "rt", encoding="ascii") + if filename.endswith((".gz", ".zip")): + with gzip_open(filename, "rt", encoding="ascii") as fp: + return json_load(fp) else: - fp = open(filename, "r") - return json_load(fp) + with open(filename, "r") as fp: + return json_load(fp) ## Method to store data into file # @@ -1022,9 +1033,9 @@ def _load_json(filename, logger: Logger = getLogger(__name__)): # @param data Data to write into file # @param filename File to write data into @staticmethod - def _store_cache_to_file(data, filename, logger: Logger = getLogger(__name__)): + def _store_cache_to_file(data, filename, logger: Logger = LOGGER): logger.debug(filename) - if filename.endswith(".gz") or filename.endswith(".zip"): + if filename.endswith((".gz", ".zip")): # Create the JSON string with sorted keys and no extra spaces # This ensures that the output is consistent for the same input data # which is important for reproducibility and caching @@ -1033,13 +1044,15 @@ def _store_cache_to_file(data, filename, logger: Logger = getLogger(__name__)): ) # fp = gzip_open(filename, 'wt', encoding='ascii', mtime=0) # Write JSON into gzip file with reproducible output - with open(filename, "wb") as raw: - with GzipFile(fileobj=raw, mode="wb", mtime=0) as f: - f.write(json_bytes) + with ( + open(filename, "wb") as raw, + GzipFile(fileobj=raw, mode="wb", mtime=0) as f, + ): + f.write(json_bytes) else: try: - fp = open(filename, "w") - json_dump(data, fp) + with open(filename, "w") as fp: + json_dump(data, fp) except FileNotFoundError as e: logger.error(str(e)) @@ -1086,7 +1099,7 @@ def _deprecatedMNX(xref_path): with open(xref_path, "rt") as f: c = csv_reader(f, delimiter="\t") for row in c: - if not row[0][0] == "#": + if row[0][0] != "#": mnx = row[0].split(":") if mnx[0] == "deprecated": deprecatedMNX_mnx[mnx[1]] = row[1] @@ -1128,11 +1141,11 @@ def _m_deprecatedMNXR(reac_xref_path): # @return cid_strc Dictionnary of formula, smiles, inchi and inchikey @staticmethod def _m_mnxm_strc( - paths: Dict[str, List[str]], + paths: dict[str, list[str]], # deprecatedCID_cid: Dict = None, interactive: bool = DEFAULTS["interactive"], - logger: Logger = getLogger(__name__), - ) -> Tuple[Dict, Dict]: + logger: Logger = LOGGER, + ) -> tuple[dict, dict]: """Parse the compounds.tsv file from RetroRules and the chem_prop.tsv file from MetanetX to generate a dictionary of compounds with their structures. Args: paths (Dict[str, List[str]]): Dictionary mapping file types to lists of paths. @@ -1164,33 +1177,31 @@ def _m_mnxm_strc( cid_name = {} for rr_compounds_path in rr_compounds_paths: - # Parse the compounds.tsv file from RetroRules - for row in csv_DictReader( - gzip_open(rr_compounds_path, "rt", encoding="utf-8-sig"), delimiter="\t" - ): - if row.get("VALID", "True").lower() != "true": - logger.debug("Skipping invalid compound entry: " + str(row)) - continue # skip invalid entries - row = { - k.lower(): v for k, v in row.items() - } # normalize keys to lowercase - if "cid" not in row: - # convert into 'id' - row["cid"] = row.pop("id") - tmp = { - "formula": row.get("formula", None), - "inchi": row.get("inchi", None), - "inchikey": row.get("inchikey", None), - "cid": row.get("cid", row.get("ID", None)), - "name": row.get("name", None), - "smiles": row.get("smiles", row.get("SMILES", None)), - } - logger.debug( - f'Processing compound {tmp["cid"]} with InChI: {tmp["inchi"]} and InChIKey: {tmp["inchikey"]}' - ) + with gzip_open(rr_compounds_path, "rt", encoding="utf-8-sig") as f: + for row in csv_DictReader(f, delimiter="\t"): + if row.get("VALID", "True").lower() != "true": + logger.debug("Skipping invalid compound entry: " + str(row)) + continue # skip invalid entries + row = { + k.lower(): v for k, v in row.items() + } # normalize keys to lowercase + if "cid" not in row: + # convert into 'id' + row["cid"] = row.pop("id") + tmp = { + "formula": row.get("formula", None), + "inchi": row.get("inchi", None), + "inchikey": row.get("inchikey", None), + "cid": row.get("cid", row.get("ID", None)), + "name": row.get("name", None), + "smiles": row.get("smiles", row.get("SMILES", None)), + } + logger.debug( + f"Processing compound {tmp['cid']} with InChI: {tmp['inchi']} and InChIKey: {tmp['inchikey']}" + ) - cid_strc[tmp["cid"]] = tmp + cid_strc[tmp["cid"]] = tmp if chem_prop_path: # Parse the chem_prop.tsv file from MetanetX @@ -1325,13 +1336,13 @@ def _m_mnxm_strc( def _m_mnxm_xref( chem_xref_path: str, # deprecatedCID_cid: Dict, - logger: Logger = getLogger(__name__), - ) -> Dict: + logger: Logger = LOGGER, + ) -> dict: cid_xref = {} with open(chem_xref_path, "rt", encoding="utf-8-sig") as f: c = csv_reader(f, delimiter="\t") for row in c: - if not row[0][0] == "#": + if row[0][0] != "#": # mnx = rrCache._checkCIDdeprecated(row[1], deprecatedCID_cid) mnx = row[1] if len(row[0].split(":")) == 1: @@ -1365,16 +1376,12 @@ def _m_mnxm_xref( # TODO: save the self.deprecatedCID_cid to be used in case there rp_paths uses an old version of MNX @staticmethod - def _m_mnxc_xref( - comp_xref_path, logger: Logger = getLogger(__name__) - ) -> Tuple[Dict, Dict]: + def _m_mnxc_xref(comp_xref_path, logger: Logger = LOGGER) -> tuple[dict, dict]: comp_xref = {} deprecatedCompID_compid = {} if not os_path.exists(comp_xref_path): - logger.error( - "Could not read the file {filename}".format(filename=comp_xref_path) - ) + logger.error(f"Could not read the file {comp_xref_path}") return None with open(comp_xref_path, "rt", encoding="utf-8-sig") as f: @@ -1382,7 +1389,7 @@ def _m_mnxc_xref( # not_recognised = [] for row in c: # cid = row[0].split(':') - if not row[0][0] == "#": + if row[0][0] != "#": # collect the info mnxc = row[1] if len(row[0].split(":")) == 1: @@ -1420,9 +1427,7 @@ def _m_mnxc_xref( # @return Dictionnary describing each reaction rule @staticmethod - def _m_rr_reactions( - rules_rall_paths: str, logger: Logger = getLogger(__name__) - ) -> Dict: + def _m_rr_reactions(rules_rall_paths: str, logger: Logger = LOGGER) -> dict: logger.debug(f"Parsing rules from {rules_rall_paths}") _rules_rall_paths = rules_rall_paths["rr2"] @@ -1430,129 +1435,124 @@ def _m_rr_reactions( rr_reactions = {} for _rules_rall_path in _rules_rall_paths: - if not os_path.exists(_rules_rall_path): logger.error( "Could not read the rules file (" + str(_rules_rall_path) + ")" ) return None - - for row in csv_DictReader( - gzip_open(_rules_rall_path, "rt"), delimiter="\t" - ): - if row["TEMPLATE_ID"] not in rr_reactions: - rr_reactions[row["TEMPLATE_ID"]] = {} - if row["REACTION_ID"] not in rr_reactions[row["TEMPLATE_ID"]]: - subtrates = {row["LEFT_IDS"]: 1} - products = dict(Counter(row["RIGHT_IDS"].split("."))) - rr_reactions[row["TEMPLATE_ID"]][row["REACTION_ID"]] = { - "rule_id": row["TEMPLATE_ID"], - "rule_score": ( - None if row["SCORE"] == "" else float(row["SCORE"]) - ), - "reac_id": row["REACTION_ID"], - "subs_id": row["LEFT_IDS"], - "rel_direction": (1 if row["DIRECTION"] == "L2R" else -1), - "left": subtrates, - "right": products, - "left_excluded": ( - row["LEFT_EXCLUDED_IDS"].split(".") - if row["LEFT_EXCLUDED_IDS"] - else [] - ), - "right_excluded": ( - row["RIGHT_EXCLUDED_IDS"].split(".") - if row["RIGHT_EXCLUDED_IDS"] - else [] - ), - } - # Handle multiple reactions per rule, update direction if needed - else: - if ( - rr_reactions[row["TEMPLATE_ID"]][row["REACTION_ID"]][ - "rel_direction" - ] - != 0 - and (1 if row["DIRECTION"] == "L2R" else -1) - != rr_reactions[row["TEMPLATE_ID"]][row["REACTION_ID"]][ - "rel_direction" - ] - ): - logger.debug( - "Updating direction for reaction " - + str(row["REACTION_ID"]) - + " in rule " - + str(row["TEMPLATE_ID"]) - + " from " - + str( - rr_reactions[row["TEMPLATE_ID"]][row["REACTION_ID"]][ - "rel_direction" - ] + with gzip_open(_rules_rall_path, "rt") as f: + for row in csv_DictReader(f, delimiter="\t"): + if row["TEMPLATE_ID"] not in rr_reactions: + rr_reactions[row["TEMPLATE_ID"]] = {} + if row["REACTION_ID"] not in rr_reactions[row["TEMPLATE_ID"]]: + subtrates = {row["LEFT_IDS"]: 1} + products = dict(Counter(row["RIGHT_IDS"].split("."))) + rr_reactions[row["TEMPLATE_ID"]][row["REACTION_ID"]] = { + "rule_id": row["TEMPLATE_ID"], + "rule_score": ( + None if row["SCORE"] == "" else float(row["SCORE"]) + ), + "reac_id": row["REACTION_ID"], + "subs_id": row["LEFT_IDS"], + "rel_direction": (1 if row["DIRECTION"] == "L2R" else -1), + "left": subtrates, + "right": products, + "left_excluded": ( + row["LEFT_EXCLUDED_IDS"].split(".") + if row["LEFT_EXCLUDED_IDS"] + else [] + ), + "right_excluded": ( + row["RIGHT_EXCLUDED_IDS"].split(".") + if row["RIGHT_EXCLUDED_IDS"] + else [] + ), + } + # Handle multiple reactions per rule, update direction if needed + else: + if ( + rr_reactions[row["TEMPLATE_ID"]][row["REACTION_ID"]][ + "rel_direction" + ] + != 0 + and (1 if row["DIRECTION"] == "L2R" else -1) + != rr_reactions[row["TEMPLATE_ID"]][row["REACTION_ID"]][ + "rel_direction" + ] + ): + logger.debug( + "Updating direction for reaction " + + str(row["REACTION_ID"]) + + " in rule " + + str(row["TEMPLATE_ID"]) + + " from " + + str( + rr_reactions[row["TEMPLATE_ID"]][ + row["REACTION_ID"] + ]["rel_direction"] + ) + + " to bidirectional (0)" ) - + " to bidirectional (0)" - ) - rr_reactions[row["TEMPLATE_ID"]][row["REACTION_ID"]][ - "rel_direction" - ] = 0 # bidirectional + rr_reactions[row["TEMPLATE_ID"]][row["REACTION_ID"]][ + "rel_direction" + ] = 0 # bidirectional return rr_reactions @staticmethod - def _m_rr_reactions_legacy( - rules_rall_path: str, logger: Logger = getLogger(__name__) - ) -> Dict: + def _m_rr_reactions_legacy(rules_rall_path: str, logger: Logger = LOGGER) -> dict: rr_reactions = {} if not os_path.exists(rules_rall_path): logger.error("Could not read the rules file (" + str(rules_rall_path) + ")") return None + with gzip_open(rules_rall_path, "rt") as f: + for row in csv_DictReader(f, delimiter="\t"): + # NOTE: as of now all the rules are generated using MNX + # but it may be that other db are used, we are handling this case + # WARNING: can have multiple products so need to seperate them + products = {} - for row in csv_DictReader(gzip_open(rules_rall_path, "rt"), delimiter="\t"): - # NOTE: as of now all the rules are generated using MNX - # but it may be that other db are used, we are handling this case - # WARNING: can have multiple products so need to seperate them - products = {} + for cid in row["Product_IDs"].split("."): + products[cid] = products.get(cid, 0) + 1 - for cid in row["Product_IDs"].split("."): - products[cid] = products.get(cid, 0) + 1 + try: + # WARNING: one reaction rule can have multiple reactions associated with them + # To change when you can set subpaths from the mutliple numbers of + # we assume that the reaction rule has multiple unique reactions associated + if row["# Rule_ID"] not in rr_reactions: + rr_reactions[row["# Rule_ID"]] = {} + if row["# Rule_ID"] in rr_reactions[row["# Rule_ID"]]: + logger.warning( + "There is already reaction " + + str(row["# Rule_ID"]) + + " in reaction rule " + + str(row["# Rule_ID"]) + ) + rr_reactions[row["# Rule_ID"]][row["Reaction_ID"]] = { + "rule_id": row["# Rule_ID"], + "rule_score": ( + None + if row["Score_normalized"] == "" + else float(row["Score_normalized"]) + ), + "reac_id": row["Reaction_ID"], + "subs_id": row["Substrate_ID"], + "rel_direction": int(row["Rule_relative_direction"]), + "left": {row["Substrate_ID"]: 1}, + "right": products, + } - try: - # WARNING: one reaction rule can have multiple reactions associated with them - # To change when you can set subpaths from the mutliple numbers of - # we assume that the reaction rule has multiple unique reactions associated - if row["# Rule_ID"] not in rr_reactions: - rr_reactions[row["# Rule_ID"]] = {} - if row["# Rule_ID"] in rr_reactions[row["# Rule_ID"]]: - logger.warning( - "There is already reaction " - + str(row["# Rule_ID"]) - + " in reaction rule " - + str(row["# Rule_ID"]) + except ValueError: + logger.error( + "Problem converting rel_direction: " + + str(row["Rule_relative_direction"]) + ) + logger.error( + "Problem converting rule_score: " + str(row["Score_normalized"]) ) - rr_reactions[row["# Rule_ID"]][row["Reaction_ID"]] = { - "rule_id": row["# Rule_ID"], - "rule_score": ( - None - if row["Score_normalized"] == "" - else float(row["Score_normalized"]) - ), - "reac_id": row["Reaction_ID"], - "subs_id": row["Substrate_ID"], - "rel_direction": int(row["Rule_relative_direction"]), - "left": {row["Substrate_ID"]: 1}, - "right": products, - } - - except ValueError: - logger.error( - "Problem converting rel_direction: " - + str(row["Rule_relative_direction"]) - ) - logger.error( - "Problem converting rule_score: " + str(row["Score_normalized"]) - ) - return rr_reactions + return rr_reactions ## Generate complete reactions from the rxn_recipes.tsv from RetroRules # @@ -1566,8 +1566,8 @@ def _m_rr_reactions_legacy( @staticmethod def _m_template_reactions( - paths: Dict[str, List[str]], logger: Logger = getLogger(__name__) - ) -> Dict: + paths: dict[str, list[str]], logger: Logger = LOGGER + ) -> dict: metadata_paths = paths["rr2"] reac_prop_path = ( @@ -1581,52 +1581,51 @@ def _m_template_reactions( # Extract reaction data from rules metadata files for metadata_path in metadata_paths: - if not os_path.exists(metadata_path): logger.error("Cannot find file: " + str(metadata_path)) return None - - for row in csv_DictReader(gzip_open(metadata_path, "rt"), delimiter="\t"): - if row["REACTION_ID"] not in reactions: - substrates = dict( - Counter( - [row["LEFT_IDS"]] - + ( - row["LEFT_EXCLUDED_IDS"].split(".") - if row["LEFT_EXCLUDED_IDS"] - else [] + with gzip_open(metadata_path, "rt") as f: + for row in csv_DictReader(f, delimiter="\t"): + if row["REACTION_ID"] not in reactions: + substrates = dict( + Counter( + [row["LEFT_IDS"]] + + ( + row["LEFT_EXCLUDED_IDS"].split(".") + if row["LEFT_EXCLUDED_IDS"] + else [] + ) ) ) - ) - products = dict( - Counter( - row["RIGHT_IDS"].split(".") - + ( - row["RIGHT_EXCLUDED_IDS"].split(".") - if row["RIGHT_EXCLUDED_IDS"] - else [] + products = dict( + Counter( + row["RIGHT_IDS"].split(".") + + ( + row["RIGHT_EXCLUDED_IDS"].split(".") + if row["RIGHT_EXCLUDED_IDS"] + else [] + ) ) ) - ) - main_left = row["LEFT_IDS"] - main_right = row["RIGHT_IDS"].split(".")[0] - if row["DIRECTION"] == "R2L": - # Swap left and right if direction is R2L - substrates, products = products, substrates - main_left, main_right = main_right, main_left - direction = -1 - else: - direction = 1 - reactions[row["REACTION_ID"]] = { - "left": substrates, - "right": products, - "direction": direction, - "main_left": main_left, - "main_right": main_right, - } - # Handle multiple reactions per rule, update direction if needed - elif row["DIRECTION"] != reactions[row["REACTION_ID"]]["direction"]: - reactions[row["REACTION_ID"]]["direction"] = 0 # bidirectional + main_left = row["LEFT_IDS"] + main_right = row["RIGHT_IDS"].split(".")[0] + if row["DIRECTION"] == "R2L": + # Swap left and right if direction is R2L + substrates, products = products, substrates + main_left, main_right = main_right, main_left + direction = -1 + else: + direction = 1 + reactions[row["REACTION_ID"]] = { + "left": substrates, + "right": products, + "direction": direction, + "main_left": main_left, + "main_right": main_right, + } + # Handle multiple reactions per rule, update direction if needed + elif row["DIRECTION"] != reactions[row["REACTION_ID"]]["direction"]: + reactions[row["REACTION_ID"]]["direction"] = 0 # bidirectional # Complete missing reactions from the reaction properties file (TSV) # Ignore all lines starting with '#', the last one contains the header '#ID' and 'mnx_equation' @@ -1654,9 +1653,7 @@ def _m_template_reactions( return reactions @staticmethod - def __load_reactions_tsv( - path: str, logger: Logger = getLogger(__name__) - ) -> "DataFrame": + def __load_reactions_tsv(path: str, logger: Logger = LOGGER) -> "DataFrame": """ Load a TSV file while: - ignoring comment lines starting with '#' @@ -1689,62 +1686,59 @@ def __load_reactions_tsv( @staticmethod def _m_template_reactions_legacy( - rxn_recipes_path: str, logger: Logger = getLogger(__name__) - ) -> Dict: + rxn_recipes_path: str, logger: Logger = LOGGER + ) -> dict: if not os_path.exists(rxn_recipes_path): logger.error("Cannot find file: " + str(rxn_recipes_path)) return None reactions = {} + with gzip_open(rxn_recipes_path, "rt") as f: + for row in csv_DictReader(f, delimiter="\t"): + # Read equation + rxn = rrCache._read_equation( + row["Equation"], row["#Reaction_ID"], logger + ) + if rxn is None: + # Pass to the next equation + continue - for row in csv_DictReader(gzip_open(rxn_recipes_path, "rt"), delimiter="\t"): - - # Read equation - rxn = rrCache._read_equation(row["Equation"], row["#Reaction_ID"], logger) - if rxn is None: - # Pass to the next equation - continue - - # Direction - dir = rrCache._read_direction(row["Direction"], logger) - if dir is None: - # Pass to the next equation - continue - else: - rxn["direction"] = dir + # Direction + dir = rrCache._read_direction(row["Direction"], logger) + if dir is None: + # Pass to the next equation + continue + else: + rxn["direction"] = dir - # Others - rxn["main_left"] = row["Main_left"].split(",") - rxn["main_right"] = row["Main_right"].split(",") + # Others + rxn["main_left"] = row["Main_left"].split(",") + rxn["main_right"] = row["Main_right"].split(",") - reactions[row["#Reaction_ID"]] = rxn + reactions[row["#Reaction_ID"]] = rxn - return reactions + return reactions - def _read_direction(dir: str, logger: Logger = getLogger(__name__)) -> Dict: + def _read_direction(dir: str, logger: Logger = LOGGER) -> dict: try: _dir = int(dir) except ValueError: ter = StreamHandler.terminator StreamHandler.terminator = "\n" - logger.warning( - "Cannot convert direction value {dir} to int".format(dir=dir) - ) + logger.warning(f"Cannot convert direction value {dir} to int") StreamHandler.terminator = ter # Pass to the next equation return None return _dir - def _read_equation( - eq: str, rxn_id: str, logger: Logger = getLogger(__name__) - ) -> Dict: + def _read_equation(eq: str, rxn_id: str, logger: Logger = LOGGER) -> dict: - if not len(eq.split("=")) == 2: + if len(eq.split("=")) != 2: logger.warning( "There should never be more or less than a left and right of an equation" ) - logger.warning("Ignoring {eq}".format(eq=eq)) + logger.warning(f"Ignoring {eq}") return None #### for character matching that are returned @@ -1820,9 +1814,10 @@ def _read_equation( # @param otype types of depiction to be generated, {"", "", ..} # @return odepic generated depictions, {"otype1": "odepic1", ..} @staticmethod - def _convert_depiction( - idepic, itype="smiles", otype={"inchikey"}, logger=getLogger(__name__) - ): + def _convert_depiction(idepic, itype="smiles", otype=None, logger: Logger = LOGGER): + if otype is None: + otype = {"inchikey"} + def MolFrom(idepic, itype, sanitize=True): if itype == "smiles": return MolFromSmiles(idepic, sanitize=sanitize) @@ -1845,7 +1840,7 @@ def MolFrom(idepic, itype, sanitize=True): f'Import error from depiction "{idepic}" of type "{itype}"' ) # Export - odepic = dict() + odepic = {} for item in otype: if item == "smiles": # MolToSmiles is tricky, one mays want to check the possible options.. diff --git a/setup.py b/setup.py index 498aadb..ffe87a7 100644 --- a/setup.py +++ b/setup.py @@ -1,6 +1,7 @@ -from setuptools import setup, find_packages from os import path as os_path +from setuptools import find_packages, setup + ## INFOS ## package = "rr_cache" descr = "Cache for RetroRules and MetaNetX" @@ -26,7 +27,7 @@ def get_version(): if line.startswith("##"): from re import search - m = search("\[(.+)\]", line) + m = search(r"\[(.+)\]", line) if m: return m.group(1) diff --git a/tests/data/compounds_rr2026.json b/tests/data/compounds_rr2026-v3.0.json similarity index 100% rename from tests/data/compounds_rr2026.json rename to tests/data/compounds_rr2026-v3.0.json diff --git a/tests/data/metrics_rr2026.json b/tests/data/metrics_rr2026-v3.0.json similarity index 100% rename from tests/data/metrics_rr2026.json rename to tests/data/metrics_rr2026-v3.0.json diff --git a/tests/data/reactions_rr2026.json b/tests/data/reactions_rr2026-v3.0.json similarity index 100% rename from tests/data/reactions_rr2026.json rename to tests/data/reactions_rr2026-v3.0.json diff --git a/tests/data/retrorules_rr2026.json b/tests/data/retrorules_rr2026-v3.0.json similarity index 100% rename from tests/data/retrorules_rr2026.json rename to tests/data/retrorules_rr2026-v3.0.json diff --git a/tests/test_rrCache.py b/tests/test_rrCache.py index 2f9ca5f..8ef5d76 100644 --- a/tests/test_rrCache.py +++ b/tests/test_rrCache.py @@ -10,17 +10,18 @@ from json import load as json_load from logging import Logger -from os import path as os_path, remove as os_rm +from os import path as os_path +from os import remove as os_rm import pytest - from brs_utils import check_file_size, create_logger, extract_gz + from rr_cache import rrCache HERE = os_path.dirname(os_path.abspath(__file__)) DATA_PATH = os_path.join(HERE, "data") -# CSPACES = ["mnx3.1", "mnx4.4", "rr2026"] -CSPACES = ["rr2026"] +# CSPACES = ["mnx3.1", "mnx4.4", "rr2026-v3.0","rr2026-v3.1.0"] +CSPACES = ["rr2026-v3.0"] DATABASES = ["metanetx", "rhea"] DATASETS = ["compounds", "metrics", "retrorules", "reactions"]