diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml deleted file mode 100644 index ff30aea..0000000 --- a/.github/workflows/check.yml +++ /dev/null @@ -1,39 +0,0 @@ -# This workflow will install Python dependencies, run tests and lint with a variety of Python versions -# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions - -name: Check - -on: - - push: - branches: [ main, master, dev ] - paths: - - '**.py' - pull_request: - branches: [ main, master ] - -jobs: - - Check: - - runs-on: ubuntu-latest - - steps: - - uses: actions/checkout@v2 - - uses: docker://continuumio/miniconda3 - - name: Updating conda - run: conda update --all -y - - name: Installing make - run: conda install -y make - - name: Checkout CI/CD Toolkit - uses: actions/checkout@v2 - with: - repository: breakthewall/cicd-toolkit - path: cicd-toolkit - persist-credentials: false - fetch-depth: 0 - - name: Checking package - run: | - cd cicd-toolkit - make check-buildenv env=check - conda run -n check make check diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml new file mode 100644 index 0000000..1445dae --- /dev/null +++ b/.github/workflows/lint.yml @@ -0,0 +1,41 @@ +name: Lint + +on: + pull_request: + push: + branches: + - main + - master + - stable + - dev + paths: + - '**.py' + +jobs: + lint: + if: github.head_ref != 'stable' + # if: github.actor != 'github-actions[bot]' + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v4 + + - name: Setup Miniconda + uses: conda-incubator/setup-miniconda@v3 + with: + activate-environment: ci-env + environment-file: environment.yaml + auto-activate-base: false + use-mamba: true + mamba-version: "*" + + - name: Add dev dependencies + shell: bash -l {0} + run: conda env update -n ci-env -f environment-dev.yaml + + - name: Run ruff + shell: bash -l {0} + run: ruff check . + - name: Run black + shell: bash -l {0} + run: black --check . diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..b76c4c7 --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,104 @@ +name: Publish + +on: + pull_request: + branches: + - stable + types: + - closed + +jobs: + Tag: + if: github.event.pull_request.merged == true + runs-on: ubuntu-latest + + outputs: + new_tag: ${{ steps.tag_version.outputs.new_tag }} + + steps: + ############################################################ + # Checkout stable branch + ############################################################ + - name: Checkout stable + uses: actions/checkout@v3 + with: + ref: stable + fetch-depth: 0 + + ############################################################ + # Create tag + ############################################################ + - name: Bump version and push tag + id: tag_version + uses: mathieudutour/github-tag-action@v5.6 + with: + github_token: ${{ secrets.GITHUB_TOKEN }} + release_branches: stable + tag_prefix: "" + + ############################################################ + # Create GitHub Release + ############################################################ + - name: Create Release + uses: actions/create-release@v1 + with: + tag_name: ${{ steps.tag_version.outputs.new_tag }} + release_name: Release ${{ steps.tag_version.outputs.new_tag }} + body: ${{ steps.tag_version.outputs.changelog }} + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + + ############################################################ + # Generate CHANGELOG + ############################################################ + - name: Setup Ruby + uses: ruby/setup-ruby@v1 + with: + ruby-version: 3 + bundler-cache: true + + - name: Generate CHANGELOG + env: + CHANGELOG_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | + gem install github_changelog_generator + AUTH=$(echo $GITHUB_REPOSITORY | sed -e 's/\// /g' | awk '{print "--user " $1 " --project " $2}') + github_changelog_generator $AUTH --no-unreleased + + ############################################################ + # Update _version.py + ############################################################ + - name: Update _version.py + env: + TAG: ${{ steps.tag_version.outputs.new_tag }} + run: | + folder=$(dirname $(find . -name _version.py)) + echo "__version__ = \"$TAG\"" > $folder/_version.py + + ############################################################ + # Commit updated files + ############################################################ + - name: Commit files + run: | + git config --local user.email "joan.herisson@univ-evry.fr" + git config --local user.name "breakthewall" + git add -A + git commit -m "chore(release): update changelog and version" || echo "No changes to commit" + + ############################################################ + # Push back to stable + ############################################################ + - name: Push changes to stable + uses: ad-m/github-push-action@master + with: + github_token: ${{ secrets.GITHUB_TOKEN }} + branch: stable + + ############################################################ + # Update main branch + ############################################################ + - name: Update main branch + uses: ad-m/github-push-action@master + with: + github_token: ${{ secrets.GITHUB_TOKEN }} + branch: main \ No newline at end of file diff --git a/.github/workflows/tag.yml b/.github/workflows/tag.yml deleted file mode 100644 index 5717cc0..0000000 --- a/.github/workflows/tag.yml +++ /dev/null @@ -1,81 +0,0 @@ -name: Tag - - -on: - push: - branches: - - stable - paths: - - '**.py' - - '!tests/**' - - -jobs: - - Tag: - # needs: [Conda-Build] - runs-on: ubuntu-latest - outputs: # Create variable usable from another job - new_tag: ${{ steps.tag_version.outputs.new_tag }} - steps: - ## CREATE TAG/RELEASE - - uses: actions/checkout@v3 - with: - persist-credentials: false # otherwise, the token used is the GITHUB_TOKEN, instead of your personal access token. - fetch-depth: 0 # otherwise, there would be errors pushing refs to the destination repository. - - name: Bump version and push tag - id: tag_version - uses: mathieudutour/github-tag-action@v5.6 - with: - github_token: ${{ secrets.GITHUB_TOKEN }} - release_branches: stable - tag_prefix: - - name: Create Release - uses: actions/create-release@v1 - with: - tag_name: ${{ steps.tag_version.outputs.new_tag }} - release_name: Release ${{ steps.tag_version.outputs.new_tag }} - body: ${{ steps.tag_version.outputs.changelog }} - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - ## UPDATE CHANGELOG - - uses: actions/checkout@v3 - with: - persist-credentials: false # otherwise, the token used is the GITHUB_TOKEN, instead of your personal access token. - fetch-depth: 0 # otherwise, there would be errors pushing refs to the destination repository. - - uses: ruby/setup-ruby@v1 - with: - ruby-version: 3 # Not needed with a .ruby-version file - bundler-cache: true # runs 'bundle install' and caches installed gems automatically - - name: Generate CHANGELOG - env: - CHANGELOG_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - run: | - gem install github_changelog_generator - AUTH=$(echo $GITHUB_REPOSITORY | sed -e 's/\// /g' | awk '{print "--user " $1 " --project " $2}') - echo $AUTH - github_changelog_generator $AUTH --no-unreleased - - name: Update _version.py - env: - TAG: ${{ steps.tag_version.outputs.new_tag }} - run: | - folder=`dirname $(find . -name _version.py)` - echo "__version__ = \"$TAG\"" > $folder/_version.py - - name: Commit files - run: | - git config --local user.email "$GITHUB_EMAIL" - git config --local user.name "$GITHUB_USERNAME" - git commit -m "doc(CHANGELOG): update" -a - env: - GITHUB_USERNAME: breakthewall - GITHUB_EMAIL: joan.herisson@univ-evry.fr - - name: Push changes - uses: ad-m/github-push-action@master - with: - github_token: ${{ secrets.GITHUB_TOKEN }} - branch: ${{ github.ref }} - - name: Update master branch - uses: ad-m/github-push-action@master - with: - github_token: ${{ secrets.GITHUB_TOKEN }} - branch: master diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 3d2f1a6..9de4c9c 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -1,39 +1,43 @@ -# This workflow will install Python dependencies, run tests and lint with a variety of Python versions -# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions - name: Tests on: - + pull_request: push: - branches: [ main, master, dev ] + branches: + - main + - master + - stable paths: - '**.py' - pull_request: - branches: [ main, master ] jobs: + test: + if: github.head_ref != 'stable' + # if: github.actor != 'github-actions[bot]' + name: Test on ${{ matrix.os }} + runs-on: ${{ matrix.os }} - Test: - - runs-on: ${{ matrix.os }}-latest strategy: + fail-fast: false matrix: - os: ["ubuntu", "macos", "windows"] - defaults: - run: - shell: bash -l {0} + os: [ubuntu-latest, macos-latest, windows-latest] steps: - - name: Checkout package repository - uses: actions/checkout@v2 - - name: Deploying miniconda - uses: conda-incubator/setup-miniconda@v2 + - uses: actions/checkout@v4 + + - name: Setup Miniconda + uses: conda-incubator/setup-miniconda@v3 with: - miniconda-version: "latest" + activate-environment: ci-env environment-file: environment.yaml - activate-environment: test - - name: Building & Testing conda package - run: | - conda install -y python pytest pytest-mock - python -m pytest + auto-activate-base: false + use-mamba: true + mamba-version: "*" + + - name: Add dev dependencies + shell: bash -l {0} + run: conda env update -n ci-env -f environment-dev.yaml + + - name: Run tests + shell: bash -l {0} + run: python -m pytest \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index adae3b5..53683d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## [3.5.0](https://github.com/brsynth/rr_cache/tree/3.5.0) (2026-02-18) + +[Full Changelog](https://github.com/brsynth/rr_cache/compare/3.4.1...3.5.0) + +## [3.4.1](https://github.com/brsynth/rr_cache/tree/3.4.1) (2025-11-04) + +[Full Changelog](https://github.com/brsynth/rr_cache/compare/3.4.0...3.4.1) + ## [3.4.0](https://github.com/brsynth/rr_cache/tree/3.4.0) (2025-10-22) [Full Changelog](https://github.com/brsynth/rr_cache/compare/3.3.2...3.4.0) 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-dev.yaml b/environment-dev.yaml new file mode 100644 index 0000000..3f43939 --- /dev/null +++ b/environment-dev.yaml @@ -0,0 +1,12 @@ +name: rr_cache +channels: + - conda-forge + - nodefaults +dependencies: + - pytest + - pytest-cov + - pytest-mock + - pytest-monitor + - black + - ruff + - mypy \ No newline at end of file diff --git a/environment.yaml b/environment.yaml index b9af508..f0d1b35 100644 --- a/environment.yaml +++ b/environment.yaml @@ -3,7 +3,8 @@ channels: - conda-forge dependencies: - python - - brs_utils>=1.22.1 + - brs_utils>=1.26.0 - requests - rdkit - colored + - pandas diff --git a/rr_cache/Args.py b/rr_cache/Args.py deleted file mode 100644 index dce2183..0000000 --- a/rr_cache/Args.py +++ /dev/null @@ -1,114 +0,0 @@ -from argparse import ArgumentParser -from typing import Callable -from os import path as os_path - -from brs_utils import add_logger_args - - -HERE = os_path.dirname(os_path.abspath( __file__ )) -CONFIG_PATH = os_path.join(HERE, 'config') -# Default values for the arguments -DEFAULTS = { - 'cspace': 'mnx3.1', - 'interactive': False, - 'do_not_dwnl_cache': False, - # default OS tmpdir, writable by the user who installs the cache, and readable by all users - 'install_dir': HERE, -} - - -def build_args_parser( - prog: str, - description: str = '', - epilog: str = '', - m_add_args: Callable = None, -) -> ArgumentParser: - - parser = ArgumentParser( - prog = prog, - description = description, - epilog = epilog - ) - - # Build Parser with rptools common arguments - parser = add_arguments(parser) - - # Add module specific arguments - if m_add_args is not None: - parser = m_add_args(parser) - - return parser - - -def add_arguments(parser: ArgumentParser) -> ArgumentParser: - # Add arguments related to the logger - parser = add_logger_args(parser) - - parser.add_argument( - '--chemical-space', - 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).' - ) - parser.add_argument( - '--list-chemical-spaces', - default=None, - action='store_true', - help='list available chemical spaces and exits' - ) - parser.add_argument( - '--build', - default=None, - action='store_true', - help='build the cache and exits. If the cache already exists, it will be overwritten. If other arguments are provided, they will be ignored.' - ) - parser.add_argument( - '--reaction-rules', - default=None, - type=str, - nargs='*', - help='print out specified reaction rule(s), all if nothing specified' - ) - parser.add_argument( - '--reactions', - default=None, - type=str, - nargs='*', - help='print out specified reaction(s), all if nothing specified' - ) - parser.add_argument( - '--compounds', - default=None, - type=str, - nargs='*', - help='print out specified compound(s), all if nothing specified' - ) - parser.add_argument( - '--chem-xref', - default=None, - type=str, - nargs='*', - help='print out specified chemical cross-reference(s), all if nothing specified' - ) - parser.add_argument( - '--interactive', - default=DEFAULTS['interactive'], - action='store_true', - dest='interactive', - help='Ask user for confirmation when loading cache (default: %(default)s)' - ) - parser.add_argument( - '--do-not-dwnl-cache', - default=DEFAULTS['do_not_dwnl_cache'], - action='store_true', - help='Do not download the cache from the remote repository' - ) - parser.add_argument( - '--install-dir', - default=DEFAULTS['install_dir'], - type=str, - help='Directory to install the cache (default: %(default)s)' - ) - - return parser diff --git a/rr_cache/__init__.py b/rr_cache/__init__.py index fdb2af4..201728a 100644 --- a/rr_cache/__init__.py +++ b/rr_cache/__init__.py @@ -1,4 +1,5 @@ -from rr_cache.rr_cache import rrCache +from .rr_cache import rrCache + # from rr_cache._version import __version__ diff --git a/rr_cache/__main__.py b/rr_cache/__main__.py index fcef812..1bd9fc4 100644 --- a/rr_cache/__main__.py +++ b/rr_cache/__main__.py @@ -1,58 +1,18 @@ -from rr_cache.rr_cache import ( - rrCache, -) -from rr_cache.Args import ( - build_args_parser -) -from logging import ( - Logger, - getLogger -) -from colored import ( - fg, - attr -) -from argparse import ( - ArgumentParser, - Namespace -) +import sys from json import dumps -from typing import ( - List, -) -from .Args import CONFIG_PATH +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 -def init( - parser: 'ArgumentParser', - args: 'Namespace' -) -> Logger: - from brs_utils import create_logger - from rr_cache._version import __version__ - - if args.log.lower() in ['silent', 'quiet'] or args.silent: - args.log = 'CRITICAL' - - if args.log.lower() in ['silent', 'quiet', 'def_info'] or args.silent: - disable_rdkit_logging() - # # Disable RDKIT logging - # from rdkit import RDLogger - # RDLogger.DisableLog('rdApp.*') - - # Create logger - logger = create_logger(parser.prog, args.log, 'rr_cache.log') - - logger.info( - '{color}{typo}rr_cache {version}{rst}{color}{rst}\n'.format( - version = __version__, - color=fg('white'), - typo=attr('bold'), - rst=attr('reset') - ) - ) - logger.debug(args) +from rr_cache.args import add_arguments +from rr_cache.rr_cache import ( + rrCache, +) - return logger +from ._version import __version__ +from .args import CONFIG_PATH def disable_rdkit_logging(): @@ -61,47 +21,52 @@ def disable_rdkit_logging(): """ import rdkit.rdBase as rkrb import rdkit.RDLogger as rkl + logger = rkl.logger() logger.setLevel(rkl.ERROR) - rkrb.DisableLog('rdApp.error') + rkrb.DisableLog("rdApp.error") def entry_point(): parser = build_args_parser( - prog = 'rr_cache', - description = 'RetroRules Cache' + prog="rr_cache", + version=__version__, + description="RetroRules Cache", + m_add_args=add_arguments, ) - args = parser.parse_args() + args = parser.parse_args() + if args.log.lower() in ["silent", "quiet", "def_info"] or args.silent: + disable_rdkit_logging() - logger = init(parser, args) + logger = init_logger(parser, args, __version__) if args.list_chemical_spaces: # list config_*.json files in CONFIG_PATH import os + files = os.listdir(CONFIG_PATH) cspaces = [] for file in files: - if file.startswith('config_') and file.endswith('.json'): - cspaces.append(file[len('config_'):-len('.json')]) + if file.startswith("config_") and file.endswith(".json"): + cspaces.append(file[len("config_") : -len(".json")]) print( - '{color}{typo}Available chemical spaces:{rst}{color}{rst}\n'.format( - color=fg('white'), - typo=attr('bold'), - rst=attr('reset') + "{color}{typo}Available chemical spaces:{rst}{color}{rst}\n".format( + color=fg("white"), typo=attr("bold"), rst=attr("reset") ) ) for dt in cspaces: - print(f'- {dt}') + print(f"- {dt}") print() - exit(0) + sys.exit(0) cache = rrCache( cspace=args.cspace, + databases=args.db, interactive=args.interactive, do_not_dwnl_cache=args.do_not_dwnl_cache, load=False, install_dir=args.install_dir, - logger=logger + logger=logger, ) # try: @@ -109,79 +74,42 @@ def entry_point(): cache.Build(interactive=args.interactive) elif args.reaction_rules is not None: print_attr( - cache, - 'rr_reactions', - args.reaction_rules, - args.do_not_dwnl_cache, - logger + cache, "rr_reactions", args.reaction_rules, args.do_not_dwnl_cache, logger ) elif args.reactions is not None: print_attr( - cache, - 'template_reactions', - args.reactions, - args.do_not_dwnl_cache, - logger + cache, "template_reactions", args.reactions, args.do_not_dwnl_cache, logger ) elif args.compounds is not None: - print_attr( - cache, - 'cid_strc', - args.compounds, - args.do_not_dwnl_cache, - logger - ) + print_attr(cache, "cid_strc", args.compounds, args.do_not_dwnl_cache, logger) elif args.chem_xref is not None: - print_attr( - cache, - 'cid_xref', - args.chem_xref, - args.do_not_dwnl_cache, - logger - ) + print_attr(cache, "cid_xref", args.chem_xref, args.do_not_dwnl_cache, logger) else: - cache.Load(interactive=args.interactive, do_not_dwnl_cache=args.do_not_dwnl_cache) - # except Exception as e: - # logger.debug(f"Exception type: {type(e).__name__}") - # logger.error( - # '\n*** An error occurred:\n{error}'.format(error=str(e)) - # ) - # logger.error('\nExiting...\n') - # exit(1) + cache.Load( + interactive=args.interactive, do_not_dwnl_cache=args.do_not_dwnl_cache + ) + + +LOGGER = getLogger(__name__) + def print_attr( - cache: 'rrCache', + 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 == []: - print( - dumps( - cache.get(attr), - indent=4 - ) - ) + print(dumps(cache.get(attr), indent=4)) else: for id in attr_lst: try: - print( - id+':', - dumps( - cache.get(attr)[id], - indent=4 - ) - ) + 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__': +if __name__ == "__main__": entry_point() diff --git a/rr_cache/_version.py b/rr_cache/_version.py index 903a158..0c11bab 100644 --- a/rr_cache/_version.py +++ b/rr_cache/_version.py @@ -1 +1 @@ -__version__ = "3.4.0" +__version__ = "3.5.1" diff --git a/rr_cache/args.py b/rr_cache/args.py new file mode 100644 index 0000000..516fc60 --- /dev/null +++ b/rr_cache/args.py @@ -0,0 +1,93 @@ +from argparse import ArgumentParser +from os import path as os_path + +HERE = os_path.dirname(os_path.abspath(__file__)) +CONFIG_PATH = os_path.join(HERE, "config") +# Default values for the arguments +DEFAULTS = { + "cspace": "rr2026-v3.1.0", + "databases": ["metanetx", "rhea"], + "interactive": False, + "do_not_dwnl_cache": False, + # default OS tmpdir, writable by the user who installs the cache, and readable by all users + "install_dir": HERE, +} + + +def add_arguments(parser: ArgumentParser) -> ArgumentParser: + + parser.add_argument( + "--chemical-space", + dest="cspace", + default=DEFAULTS["cspace"], + type=str, + 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", + default=None, + action="store_true", + help="list available chemical spaces and exits", + ) + parser.add_argument( + "--db", + "--databases", + default=DEFAULTS["databases"], + nargs="*", + help="select databases to include in the cache (default: %(default)s)", + ) + parser.add_argument( + "--build", + default=None, + action="store_true", + help="build the cache and exits. If the cache already exists, it will be overwritten. If other arguments are provided, they will be ignored.", + ) + parser.add_argument( + "--reaction-rules", + default=None, + type=str, + nargs="*", + help="print out specified reaction rule(s), all if nothing specified", + ) + parser.add_argument( + "--reactions", + default=None, + type=str, + nargs="*", + help="print out specified reaction(s), all if nothing specified", + ) + parser.add_argument( + "--compounds", + default=None, + type=str, + nargs="*", + help="print out specified compound(s), all if nothing specified", + ) + parser.add_argument( + "--chem-xref", + default=None, + type=str, + nargs="*", + help="print out specified chemical cross-reference(s), all if nothing specified", + ) + parser.add_argument( + "--interactive", + default=DEFAULTS["interactive"], + action="store_true", + dest="interactive", + help="Ask user for confirmation when loading cache (default: %(default)s)", + ) + parser.add_argument( + "--do-not-dwnl-cache", + default=DEFAULTS["do_not_dwnl_cache"], + action="store_true", + help="Do not download the cache from the remote repository", + ) + parser.add_argument( + "--install-dir", + default=DEFAULTS["install_dir"], + type=str, + help="Directory to install the cache (default: %(default)s)", + ) + + return parser diff --git a/rr_cache/config/config_mnx3.1.json b/rr_cache/config/config_mnx3.1.json index 72a0aac..22d14e3 100644 --- a/rr_cache/config/config_mnx3.1.json +++ b/rr_cache/config/config_mnx3.1.json @@ -9,18 +9,18 @@ "file": { "url": "https://gitlab.com/breakthewall/rrCache-data/-/raw/master/cache/mnx_3.1/", "name": "cid_strc.json.gz", - "fingerprint": "ecd2111d08694cc64702991774722ecb110f82cb310969ae5c4844fd8d2a39340084eeca7d6e409b718594ec998d19f63c2129657bb8299a2799d321750dfc91" + "fingerprint": "5b44f1d9e5b7e00f460f135e9539272cd8cea302baa7fb08c22e914be642fd4543525c934dec0386562397d0c767a2565eac1f1711a5e838e0aab7aba43b5166" } }, "cid_xref": { "deps": { "attr_deps": ["deprecatedCID_cid"], - "file_deps": [] + "file_deps": ["chem_xref.tsv"] }, "file": { "url": "https://gitlab.com/breakthewall/rrCache-data/-/raw/master/cache/mnx_3.1/", "name": "cid_xref.json.gz", - "fingerprint": "eb1c9aee49b1fdbe873c7d7f5c6af5f706ff06476c5e69c6c0fa5350c2b516e589881874b8139b091e6cad5dbe363c2a052cbcdfe550e490d3bd87da1b2fabd4" + "fingerprint": "ff25b26e13cb100f341962877724e68f7f41c5928d317096149277d0776c493d23cd7eb4b20e3d81f964ac51e79a7f6c8261d4098a12855eaeea3bf76d93c9a2" } }, "chebi_cid": { @@ -42,7 +42,7 @@ "file": { "url": "https://gitlab.com/breakthewall/rrCache-data/-/raw/master/cache/mnx_3.1/", "name": "rr_reactions.json.gz", - "fingerprint": "1ef37066d168307429eff4901cfc4db82795d13879260204f2b32399c9b0621fb0d934322a3fd3bde145b0273b76e4d7c1bb34d641b90165403308bde68fddef" + "fingerprint": "4325233ce6373de6aa01e86eb16e78c234a33180f15658213d8fd9d07f0b6f4073139a82bf132482e0a659982d091f38c84bfd66d12b0cf4917acb1177a1a64e" } }, "inchikey_cid": { @@ -108,7 +108,7 @@ "file": { "url": "https://gitlab.com/breakthewall/rrCache-data/-/raw/master/cache/mnx_3.1/", "name": "template_reactions.json.gz", - "fingerprint": "dd626073f06f3cd91352eed111080b9ece4ca67cc954917f7377e537fb99b11d587ccfbc1f9510627fe0a50f5ca5eccbf881af6d152764ed2df498811d7fbcb2" + "fingerprint": "0a4c9d10e59132652a0aee07648a1cd436259399168b6f04e0190df9b10fdc44b92a1106f761834c9ee1ba0cdc4902631e384d4bbe93ef5361f6623da68d630e" } } }, diff --git a/rr_cache/config/config_mnx4.4.json b/rr_cache/config/config_mnx4.4.json index 0e593df..e16e4f1 100644 --- a/rr_cache/config/config_mnx4.4.json +++ b/rr_cache/config/config_mnx4.4.json @@ -31,7 +31,7 @@ "file": { "url": "https://gitlab.com/breakthewall/rrCache-data/-/raw/master/cache/mnx_4.4/", "name": "cid_xref.json.gz", - "fingerprint": "f8a6e6ceb529abcb0223f89a92692a5031e31e2a7b35fa3f5e0b741b822544caa74b7cdc1ea137239fa6d5af81ad1d2794283157996185a20a23eed8a921b66a" + "fingerprint": "a81c5217769bb1d83ebe8d1905cfec453f23837ad85c22a30710629d434ffcc878431cc3cc7f48a3cf70f9cf45647361a9e45cd89dca7403d1254663a82de6bc" } }, "template_reactions": { diff --git a/rr_cache/config/config_rr2026-v3.0.json b/rr_cache/config/config_rr2026-v3.0.json new file mode 100644 index 0000000..d43945d --- /dev/null +++ b/rr_cache/config/config_rr2026-v3.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.0/", + "files": { + "templates_metadata": { + "metanetx": "a35395aea2fb31ac1dce3f790b47c0d22508bca80f982d56c6f053133d42b322b21b6d910a869f2fe55d749fed8d650eff1f101f8386cfc28dd395baca9f798a", + "rhea": "5b966e1b71abdc7562cdf64a960e9892e5599ed2b6e6e4a70277dfe3af482d9a12246037ff452bcec20436d58234d4a3ab2a11e188acec1f2508875ae7658000", + "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/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/config/config_rr2026.json b/rr_cache/config/config_rr2026.json deleted file mode 100644 index 785b049..0000000 --- a/rr_cache/config/config_rr2026.json +++ /dev/null @@ -1,49 +0,0 @@ -{ - "type": "rr2026", - "cache": { - "cid_strc": { - "deps": { - "file_deps": ["chemicals.tsv.gz", "chem_prop.tsv"] - }, - "file": { - "url": "", - "name": "cid_strc.json.gz", - "fingerprint": "4d51044442db3a103b0540df3dd24bf946bc0f13ef0f561f5baaaaa19fd57fd450885971c7b1661f3fa410250dd86c186c899c36891ef3f86eeef51181620386" - } - }, - "rr_reactions": { - "deps": { - "file_deps": ["templates.tsv.gz"] - }, - "file": { - "url": "", - "name": "rr_reactions.json.gz", - "fingerprint": "b6e2dbc7033bafd21e6567cef8f75de06cf7e7bd365e361d8aff346df159aeaae3ba56c6d1d543cd985587c81e3a11d3931b1d1896214d8485cc44bbf54266d1" - } - }, - "template_reactions": { - "deps": { - "file_deps": ["templates.tsv.gz"] - }, - "file": { - "url": "", - "name": "template_reactions.json.gz", - "fingerprint": "78946fd50963c1587e43caf190a3853936b3ed18bd2e649447b7e9c936f530e52bcc2d91ae6c97cae5fb58bde47891f2e1cc85cc7fe8e82d40a03fa6e0624d50" - } - } - }, - "sources": { - "rr2": { - "url": "", - "files": { - "templates.tsv.gz": "ede287e7e2b08f5620d26af0d7e68e25be70fb54eb5a8657b92314d4250a869073fa6022608fb00b9147a531a81f32c31923613a9b2b0c4321da45a7d8f56a1b" - } - }, - "rr2more": { - "url": "", - "files": { - "chemicals.tsv.gz": "af060a170e43e55b0ee70567d7f14ef9ae21534c7c5c2e6a58ec19f8adec82a6c05080204c1d6d43f1556cabf91b711ba85d58dc5a6b1b5cb5e892fde4d7e001" - } - } - } -} \ No newline at end of file diff --git a/rr_cache/rr_cache.py b/rr_cache/rr_cache.py index cb76cb5..a09d559 100644 --- a/rr_cache/rr_cache.py +++ b/rr_cache/rr_cache.py @@ -1,69 +1,52 @@ -from os import ( - path as os_path, - rename as os_rename, - makedirs +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 +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 brs_utils import check_sha, download, print_end, print_progress, print_start +from colored import ( + attr as c_attr, +) +from pandas import ( + DataFrame, +) +from pandas import ( + read_csv as pd_read_csv, ) -from tempfile import NamedTemporaryFile from rdkit.Chem import ( - MolFromSmiles, MolFromInchi, - MolToSmiles, + MolFromSmiles, MolToInchi, MolToInchiKey, + MolToSmiles, ) -from csv import ( - DictReader as csv_DictReader, - reader as csv_reader -) -from json import ( - dump as json_dump, - dumps as json_dumps, - load as json_load -) -from gzip import ( - open as gzip_open, - GzipFile -) -from re import findall as re_findall -# from time import time as time_time -from requests import exceptions as r_exceptions -from hashlib import sha512 -from pathlib import Path -from colored import ( - attr as c_attr, -) -from logging import ( - Logger, - getLogger, - StreamHandler -) -from typing import ( - List, - Tuple, - Dict -) -from collections import ( - defaultdict, - Counter -) -from brs_utils import ( - print_start, - print_progress, - print_end, - download, - check_sha -) -from .Args import DEFAULTS, HERE, CONFIG_PATH +from requests import exceptions as r_exceptions + +from .args import CONFIG_PATH, DEFAULTS def ask_user_input(): # Ask the user for input - user_input = input("Press Enter to continue, 's' to stop, or 'c' to continue without prompting: ") + user_input = input( + "Press Enter to continue, 's' to stop, or 'c' to continue without prompting: " + ) # Check the user's input - if user_input.lower() == 's': + if user_input.lower() == "s": print("Stopping the program.") - exit() - elif user_input.lower() == 'c': + sys.exit() + elif user_input.lower() == "c": print("Continuing without prompting.") return False else: @@ -74,10 +57,14 @@ def ask_user_input(): class FileCorruptedError(Exception): pass + class FingerprintError(Exception): pass +LOGGER = getLogger(__name__) + + class rrCache: """Class to generate the cache @@ -87,20 +74,21 @@ class rrCache: # __input__cache_url = 'ftp://ftp.vital-it.ch/databases/metanetx/MNXref/3.2/' - ## Cache constructor def __init__( self, - cspace: str = DEFAULTS['cspace'], - interactive: bool = DEFAULTS['interactive'], - do_not_dwnl_cache: bool = DEFAULTS['do_not_dwnl_cache'], + cspace: str = DEFAULTS["cspace"], + 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__) - ) -> 'rrCache': + install_dir: str = DEFAULTS["install_dir"], + logger: Logger = LOGGER, + ) -> "rrCache": """Constructor for the class Args: cspace (str): Chemical space to use (e.g. mnx3.1, mnx4.4...). + databases (List[str]): List of databases to include in the cache. interactive (bool): Whether to ask the user for confirmation before overwriting existing files. do_not_dwnl_cache (bool): Whether to download the cache files from the internet. load (bool): Whether to load the cache files into memory. @@ -109,65 +97,71 @@ def __init__( """ self.logger = logger - self.logger.debug('New instance of rrCache') - self.logger.debug('cspace: '+str(cspace)) - self.logger.debug('interactive: '+str(interactive)) - self.logger.debug('do_not_dwnl_cache: '+str(do_not_dwnl_cache)) - self.logger.debug('load: '+str(load)) - self.logger.debug('install_dir: '+str(install_dir)) + self.logger.debug("New instance of rrCache") + self.logger.debug("cspace: " + str(cspace)) + self.logger.debug("databases: " + str(databases)) + self.logger.debug("interactive: " + str(interactive)) + self.logger.debug("do_not_dwnl_cache: " + str(do_not_dwnl_cache)) + self.logger.debug("load: " + str(load)) + self.logger.debug("install_dir: " + str(install_dir)) self.__cspace = cspace # Config file # Search first in install_dir, # then in default config path - cache_cfg_fln = os_path.join(install_dir, 'config', f'config_{self.__cspace}.json') + cache_cfg_fln = os_path.join( + install_dir, "config", f"config_{self.__cspace}.json" + ) if not os_path.exists(cache_cfg_fln): - cache_cfg_fln = os_path.join(CONFIG_PATH, f'config_{self.__cspace}.json') + cache_cfg_fln = os_path.join(CONFIG_PATH, f"config_{self.__cspace}.json") try: - with open(cache_cfg_fln, 'r') as f: + with open(cache_cfg_fln, "r") as f: cache_cfg = json_load(f) - rrCache.__cache_sources = cache_cfg['sources'] - rrCache.__cache = cache_cfg['cache'] - rrCache.__type = cache_cfg.get('type', 'legacy') + rrCache.__cache_sources = cache_cfg["sources"] + rrCache.__cache = cache_cfg["cache"] + rrCache.__type = cache_cfg.get("type", "legacy") + rrCache.__databases = databases except FileNotFoundError: - logger.error(f'Cache config file {cache_cfg_fln} not found, please check the --chemical-space argument') - logger.error('Exiting...') - exit(1) - + logger.error( + f"Cache config file {cache_cfg_fln} not found, please check the --chemical-space argument" + ) + logger.error("Exiting...") + sys.exit(1) # Cache elements list rrCache.__attributes_list = list(rrCache.__cache.keys()) # static attribues - if 'static' in cache_cfg: + if "static" in cache_cfg: try: - convert_fln = os_path.join(CONFIG_PATH, cache_cfg['static']) - with open(convert_fln, 'r') as f: + convert_fln = os_path.join(CONFIG_PATH, cache_cfg["static"]) + with open(convert_fln, "r") as f: rrCache.__convertMNXM = json_load(f) except FileNotFoundError: rrCache.__convertMNXM = {} - self.logger.info(f'Using {self.__cspace}') - self.__input__cache_dir = os_path.join(install_dir, 'input-cache', self.__cspace) - self.__cache_dir = os_path.join(install_dir, 'cache', self.__cspace) + self.logger.info(f"Using {self.__cspace}") + self.__input__cache_dir = os_path.join( + install_dir, "input-cache", self.__cspace + ) + self.__cache_dir = os_path.join(install_dir, "cache", self.__cspace) if load: self.Load( attrs=rrCache.__attributes_list, interactive=interactive, do_not_dwnl_cache=do_not_dwnl_cache, - type=rrCache.__type + type=rrCache.__type, ) - def get_cspace(self) -> str: """Get the chemical space used Returns: str: Chemical space """ return self.__cspace - + def get_type(self) -> str: """Get the chemical space type used Returns: @@ -177,10 +171,10 @@ def get_type(self) -> str: def Load( self, - attrs: List = [], - interactive: bool = DEFAULTS['interactive'], - do_not_dwnl_cache: bool = DEFAULTS['do_not_dwnl_cache'], - type: str = 'legacy' + attrs: list | None = None, + interactive: bool = DEFAULTS["interactive"], + do_not_dwnl_cache: bool = DEFAULTS["do_not_dwnl_cache"], + type: str = "legacy", ) -> None: """Load the cache attributes into memory Args: @@ -189,44 +183,45 @@ def Load( Returns: None """ - self.logger.debug('Loading attributes: '+str(attrs)) - self.logger.debug('interactive: '+str(interactive)) - self.logger.debug('do_not_dwnl_cache: '+str(do_not_dwnl_cache)) + self.logger.debug("Loading attributes: " + str(attrs)) + self.logger.debug("interactive: " + str(interactive)) + 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): - self.logger.warning('\'attrs\' argument is not of type list, trying to convert...') + self.logger.warning( + "'attrs' argument is not of type list, trying to convert..." + ) self.__attributes_list = [attrs] - self.logger.warning('OK') + self.logger.warning("OK") else: self.__attributes_list = attrs for attr in self.__attributes_list: - setattr(self, '__'+attr, None) + setattr(self, "__" + attr, None) if not do_not_dwnl_cache: try: rrCache._check_or_download_cache_to_disk( - self.__cache_dir, - self.__attributes_list, - self.logger + self.__cache_dir, self.__attributes_list, self.logger ) except r_exceptions.MissingSchema as e: self.logger.warning(e) try: self._check_or_load_cache() - except (FileNotFoundError, - r_exceptions.RequestException, - r_exceptions.InvalidSchema, - r_exceptions.ConnectionError): + except ( + FileNotFoundError, + r_exceptions.RequestException, + r_exceptions.InvalidSchema, + r_exceptions.ConnectionError, + ): self.Build(interactive=interactive) self._check_or_load_cache() - def get_input_cache_dir(self) -> str: """Get the input cache directory Returns: @@ -236,51 +231,49 @@ 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)) + logger.debug("cache_dir: " + str(cache_dir)) + logger.debug("attributes: " + str(attributes_list)) # print_start(logger, 'Downloading cache') for attr in attributes_list: print_progress() - filename = rrCache.__cache[attr]['file']['name'] + filename = rrCache.__cache[attr]["file"]["name"] full_filename = os_path.join(cache_dir, filename) # try: if os_path.exists(full_filename): - fingerprint = rrCache.__cache[attr]['file']['fingerprint'] - if check_sha( - full_filename, - fingerprint - ): - logger.debug(filename+" already downloaded") - else: # sha not ok - logger.warning( - '\nfilename: ' + filename - + '\nlocation: ' + cache_dir - + '\nsha (computed): ' + sha512(Path(full_filename).read_bytes()).hexdigest() - + '\nsha (expected): ' + fingerprint - ) - # raise FileNotFoundError + # fingerprint = rrCache.__cache[attr]["file"]["fingerprint"] + # if check_sha(full_filename, fingerprint): + logger.debug(filename + " already downloaded") + # else: # sha not ok + # logger.warning( + # "\nfilename: " + # + filename + # + "\nlocation: " + # + cache_dir + # + "\nsha (computed): " + # + sha512(Path(full_filename).read_bytes()).hexdigest() + # + "\nsha (expected): " + # + fingerprint + # ) + # # raise FileNotFoundError else: - # raise FileNotFoundError + # raise FileNotFoundError - # except FileNotFoundError: - logger.debug("Downloading "+filename+"...") + # except FileNotFoundError: + logger.debug("Downloading " + filename + "...") # start_time = time_time() if not os_path.isdir(cache_dir): # cache_dir = '/mnx4.4' makedirs(cache_dir, exist_ok=True) - if rrCache.__cache[attr]['file']['url'] == "": + if rrCache.__cache[attr]["file"]["url"] == "": raise r_exceptions.MissingSchema( - f'URL for {attr} is empty, cannot download cache file' + f"URL for {attr} is empty, cannot download cache file" ) else: download( - rrCache.__cache[attr]['file']['url']+filename, - full_filename + rrCache.__cache[attr]["file"]["url"] + filename, full_filename ) # rrCache.__cache[attr] = True # end_time = time_time() @@ -294,42 +287,116 @@ def _check_or_download_cache_to_disk( def get(self, attr: str): # self.logger.debug(f'Getting attribute: {attr}') try: - return getattr(self, '__'+attr) - except AttributeError: # Try class attribute + return getattr(self, "__" + attr) + except AttributeError: # Try class attribute 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 def __hasattr(self, attr: str): - return hasattr(self, '__'+attr) + return hasattr(self, "__" + attr) def get_compound(self, cid: str): - return self.__get_object('cid_strc', cid) + return self.__get_object("cid_strc", cid) + + def get_compound_xref(self, cid: str): + return self.__get_object("cid_xref", cid) def get_list_of_compounds(self): - return self.__get_list_of_objects('cid_strc') + return self.__get_list_of_objects("cid_strc") def get_reaction(self, rxn_id: str): - return self.__get_object('template_reactions', rxn_id) + return self.__get_object("template_reactions", rxn_id) + + @staticmethod + def _m_mnx_reaction_from_reac_prop( + reac_prop_path: str, + rxn_id: str, + 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 + + with open(reac_prop_path, "rt", encoding="utf-8-sig") as f: + reader = csv_reader(f, delimiter="\t") + header = None + for row in reader: + if not row or len(row) == 0: + continue + if row[0].startswith("#ID"): + header = [h.lstrip("#") for h in row] + continue + if row[0].startswith("#"): + continue + if header is None: + continue + + row_dict = { + header[i]: row[i] if i < len(row) else "" + for i in range(len(header)) + } + if row_dict.get("ID") != rxn_id: + continue + + equation = row_dict.get("mnx_equation") or row_dict.get("equation") + if not equation: + logger.warning( + f"No equation found for reaction {rxn_id} in reac_prop.tsv" + ) + return None + + parsed = rrCache._read_equation(equation, rxn_id, logger) + if parsed is None: + return None + + left = parsed.get("left", {}) + right = parsed.get("right", {}) + main_left = [next(iter(left.keys()))] if left else [] + main_right = [next(iter(right.keys()))] if right else [] + + return { + "left": left, + "right": right, + "direction": 0, + "main_left": main_left, + "main_right": main_right, + } + + return None def get_list_of_reactions(self): - return self.__get_list_of_objects('template_reactions') + return self.__get_list_of_objects("template_reactions") def get_reaction_rule(self, rr_id: str): - return self.__get_object('rr_reactions', rr_id) + return self.__get_object("rr_reactions", rr_id) def get_list_of_reaction_rules(self): - return self.__get_list_of_objects('rr_reactions') + return self.__get_list_of_objects("rr_reactions") + + def __contains__(self, id: str) -> bool: + if not isinstance(id, str): + return False + + for attr in ("cid_strc", "template_reactions", "rr_reactions"): + try: + if not self.__hasattr(attr): + self.Load(attrs=[attr]) + if id in self.get(attr): + return True + except AttributeError: + continue + return False def __get_object(self, attr: str, id: str): try: 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): @@ -337,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: + return setattr(self, "__" + attr, val) + except AttributeError as e: self.logger.error(str(e)) ##################################################### @@ -351,14 +418,11 @@ def set(self, attr: str, val: object): ##################################################### class Error(Exception): - """Error function for the convertion of structures - """ - pass - + """Error function for the convertion of structures""" class DepictionError(Error): - """Error function for the convertion of structures - """ + """Error function for the convertion of structures""" + def __init__(self, message): """Constructor for the class @@ -372,406 +436,485 @@ def __init__(self, message): # self.expression = expression self.message = message -# @staticmethod - def Build( - self, - interactive: bool = DEFAULTS['interactive'] + @staticmethod + def _download_if_not_exists_or_corrupted( + url: str, + filename: str, + outdir: str, + fingerprint: str, + logger: Logger = LOGGER, ) -> None: + # Download if not exists or corrupted + if not os_path.exists(os_path.join(outdir, filename)): + logger.debug(f"{filename} not found in {outdir}, downloading...") + logger.debug(f"Trying to download from {url}...") + try: + rrCache._download_input_cache( + url=url, + file=filename, + outdir=outdir, + fingerprint=fingerprint, + logger=logger, + ) + except r_exceptions.RequestException as e: + logger.warning(f"Failed to download {filename} from {url}: {e}") + else: + if not check_sha( + os_path.join(outdir, filename), + {"file": {"fingerprint": fingerprint}}, + ): + logger.debug( + f"{filename} found in input cache but corrupted, re-downloading..." + ) + rrCache._download_input_cache( + url=url, + file=filename, + outdir=outdir, + fingerprint=fingerprint, + logger=logger, + ) + else: + logger.debug( + f"{filename} found in input cache and valid, skipping download" + ) + + def Build(self, interactive: bool = DEFAULTS["interactive"]) -> None: """Generate the cache files and store them to disk. Args: interactive (bool): Whether to ask the user for confirmation before overwriting existing files. logger (Logger): Logger instance for logging messages. """ - self.logger.debug('interactive: '+str(interactive)) + self.logger.debug("interactive: " + str(interactive)) # CACHE if os_path.isdir(self.__cache_dir): - self.logger.warning(f'Cache directory {self.__cache_dir} already exists, data might be overwritten') + self.logger.warning( + f"Cache directory {self.__cache_dir} already exists, data might be overwritten" + ) else: makedirs(self.__cache_dir) # INPUT_CACHE if os_path.isdir(self.__input__cache_dir): - self.logger.warning(f'Input cache directory {self.__input__cache_dir} already exists, existing data can be used, remove the directory to download new data') + self.logger.warning( + f"Input cache directory {self.__input__cache_dir} already exists, existing data can be used, remove the directory to download new data" + ) else: makedirs(self.__input__cache_dir) # FETCH INPUT_CACHE FILES - print_start(self.logger, 'Checking input cache') + 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(): - # filename = 'metanetx' + os_path.sep + filename - # Download if not exists or corrupted - if not os_path.exists(os_path.join(self.__input__cache_dir, filename)): - self.logger.debug(f'{filename} not found in {self.__input__cache_dir}, downloading...') - rrCache._download_input_cache( - url=input['url'], - file=filename, + 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() + for db in databases: + if db not in self.__databases: + self.logger.debug( + f"Database {db} is not in the list of databases to include in the cache, skipping download of {filename} for {db}" + ) + continue + url = input["url"] + db + "/" + outdir = os_path.join(self.__input__cache_dir, db) + if db in fingerprint: + rrCache._download_if_not_exists_or_corrupted( + url=url, + filename=filename, + outdir=outdir, + fingerprint=fingerprint[db], + logger=self.logger, + ) + else: + self.logger.warning( + f"Database {db} not found in fingerprint configuration for {input_type}, skipping download of {filename} for {db}" + ) + + else: # one single database-independent file to download + rrCache._download_if_not_exists_or_corrupted( + url=input["url"], + filename=filename, outdir=self.__input__cache_dir, fingerprint=fingerprint, - logger=self.logger + logger=self.logger, ) - else: - if not check_sha( - os_path.join(self.__input__cache_dir, filename), - { - 'file': { - 'fingerprint': fingerprint - } - } - ): - self.logger.debug(f'{filename} found in input cache but corrupted, re-downloading...') - rrCache._download_input_cache( - url=input['url'], - file=filename, - outdir=self.__input__cache_dir, - fingerprint=fingerprint, - logger=self.logger - ) - else: - self.logger.debug(f'{filename} found in input cache and valid, skipping download') - print_end(self.logger) + # BUILD CACHE FILES AND STORE THEM TO DISK - print_start(self.logger, 'Building the cache') - # try: - # deprecatedCID_cid = rrCache._gen_deprecatedCID_cid(self.__input__cache_dir, self.__cache_dir, self.logger) - # except KeyError as e: - # self.logger.debug(f'{e} not found in input cache, skipping generation') - # deprecatedCID_cid = None - # print_progress(self.logger) - cid_strc, cid_name = rrCache._gen_cid_strc_cid_name(self.__input__cache_dir, self.__cache_dir, interactive=interactive, type=rrCache.__type, logger=self.logger) + print_start(self.logger, "Building the cache") + cid_strc, cid_name = rrCache._gen_cid_strc_cid_name( + self.__input__cache_dir, + self.__cache_dir, + interactive=interactive, + type=rrCache.__type, + databases=self.__databases, + logger=self.logger, + ) print_progress(self.logger) try: - rrCache._gen_inchikey_cid(self.__input__cache_dir, self.__cache_dir, cid_strc, self.logger) + rrCache._gen_inchikey_cid( + self.__input__cache_dir, self.__cache_dir, cid_strc, self.logger + ) except KeyError as e: - self.logger.debug(f'{e} not found in input cache, skipping generation') + self.logger.debug(f"{e} not found in input cache, skipping generation") print_progress(self.logger) del cid_strc, cid_name try: - cid_xref = rrCache._gen_cid_xref(self.__input__cache_dir, self.__cache_dir, self.logger) - print_progress(self.logger) - rrCache._gen_chebi_cid(self.__input__cache_dir, self.__cache_dir, cid_xref) + cid_xref = rrCache._gen_cid_xref( + self.__input__cache_dir, self.__cache_dir, logger=self.logger + ) print_progress(self.logger) + # rrCache._gen_chebi_cid(self.__input__cache_dir, self.__cache_dir, cid_xref) + # print_progress(self.logger) del cid_xref except KeyError as e: - self.logger.debug(f'{e} not found in input cache, skipping generation') + self.logger.debug(f"{e} not found in input cache, skipping generation") # try: # deprecatedRID_rid = rrCache._gen_deprecatedRID_rid(self.__input__cache_dir, self.__cache_dir, self.logger) # except KeyError as e: # deprecatedRID_rid = None # self.logger.debug(f'{e} not found in input cache, skipping generation') # print_progress(self.logger) - rrCache._gen_rr_reactions(self.__input__cache_dir, self.__cache_dir, type=rrCache.__type, logger=self.logger) # , deprecatedCID_cid, deprecatedRID_rid, logger) - print_progress(self.logger) + for attribute in ["template", "rr"]: + rrCache._gen_reactions( + self.__input__cache_dir, + self.__cache_dir, + type=rrCache.__type, + databases=self.__databases, + attribute=attribute, + logger=self.logger, + ) # , deprecatedCID_cid, deprecatedRID_rid, logger) + print_progress(self.logger) try: - rrCache._gen_comp_xref_deprecatedCompID_compid(self.__input__cache_dir, self.__cache_dir, self.logger) + rrCache._gen_comp_xref_deprecatedCompID_compid( + self.__input__cache_dir, self.__cache_dir, self.logger + ) except KeyError as e: - self.logger.debug(f'{e} not found in input cache, skipping generation') - print_progress(self.logger) - rrCache._gen_template_reactions(self.__input__cache_dir, self.__cache_dir, type=rrCache.__type, logger=self.logger) # , deprecatedCID_cid, deprecatedRID_rid, logger) + self.logger.debug(f"{e} not found in input cache, skipping generation") print_progress(self.logger) + # rrCache._gen_reactions( + # self.__input__cache_dir, + # self.__cache_dir, + # type=rrCache.__type, + # databases=self.__databases, + # attribute="template", + # logger=self.logger, + # ) # , deprecatedCID_cid, deprecatedRID_rid, logger) + # print_progress(self.logger) # del deprecatedCID_cid, deprecatedRID_rid print_progress(self.logger) print_end(self.logger) @staticmethod def _gen_deprecatedCID_cid( - input_dir: str, - outdir: str, - logger: Logger = getLogger(__name__) - ) -> Dict: - attribute = 'deprecatedCID_cid' - logger.debug(c_attr('bold')+attribute+c_attr('reset')) + 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}") logger.debug(f" outdir: {outdir}") deprecatedCID_cid = None - f_deprecatedCID_cid = os_path.join(outdir, rrCache.__cache[attribute]['file']['name']) + f_deprecatedCID_cid = os_path.join( + outdir, rrCache.__cache[attribute]["file"]["name"] + ) logger.debug(f" f_deprecatedCID_cid: {f_deprecatedCID_cid}") # Do not checksum since it is a dictionary if os_path.exists(f_deprecatedCID_cid) and check_sha( - f_deprecatedCID_cid, - rrCache.__cache[attribute] + f_deprecatedCID_cid, rrCache.__cache[attribute] ): deprecatedCID_cid = rrCache._load_json(f_deprecatedCID_cid) logger.debug(" Cache file already exists") else: logger.debug(" Generating data...") deprecatedCID_cid = rrCache._m_deprecatedMNXM( - os_path.join(input_dir, 'chem_xref.tsv') + os_path.join(input_dir, "chem_xref.tsv") ) logger.debug(" Writing data to file...") - rrCache._store_cache_to_file(deprecatedCID_cid, f_deprecatedCID_cid, logger=logger) - - return { - 'attr': deprecatedCID_cid, - 'file': f_deprecatedCID_cid - } + rrCache._store_cache_to_file( + deprecatedCID_cid, f_deprecatedCID_cid, logger=logger + ) + return {"attr": deprecatedCID_cid, "file": f_deprecatedCID_cid} @staticmethod def _gen_cid_strc_cid_name( input_dir: str, outdir: str, - interactive: bool = DEFAULTS['interactive'], - type: str = 'legacy', - logger: Logger = getLogger(__name__) - ) -> Dict: - - attribute = 'cid_strc, cid_name' - logger.debug(c_attr('bold')+attribute+c_attr('reset')) + interactive: bool = DEFAULTS["interactive"], + type: str = "legacy", + databases: list[str] = DEFAULTS["databases"], + logger: Logger = LOGGER, + ) -> dict: + + logger.debug(f"input_dir: {input_dir}") + logger.debug(f"outdir: {outdir}") + logger.debug(f"interactive: {interactive}") + logger.debug(f"type: {type}") + logger.debug(f"databases: {databases}") + + attribute = "cid_strc, cid_name" + logger.debug(c_attr("bold") + attribute + c_attr("reset")) cid_strc = None - # cid_name = None - f_cid_strc = os_path.join(outdir, rrCache.__cache['cid_strc']['file']['name']) - # try: - # f_cid_name = os_path.join(outdir, rrCache.__cache['cid_name']['file']['name']) - # except KeyError: - # f_cid_name = "" - # logger.debug(" No cid_name file found in cache, skipping generation of cid_name") + f_cid_strc = os_path.join(outdir, rrCache.__cache["cid_strc"]["file"]["name"]) # Do not checksum since it is a dictionary if os_path.exists(f_cid_strc) and check_sha( - f_cid_strc, - rrCache.__cache['cid_strc'] - ): #and os_path.exists(f_cid_name) and check_sha( - # f_cid_name, - # rrCache.__cache['cid_name'] - # ): + f_cid_strc, rrCache.__cache["cid_strc"] + ): cid_strc = rrCache._load_json(f_cid_strc) logger.debug(" Cache file already exists") else: - # if deprecatedCID_cid: - # if not deprecatedCID_cid['attr']: - # logger.debug(" Loading input data from file...") - # deprecatedCID_cid = rrCache._load_json(deprecatedCID_cid['file']) - # else: - # deprecatedCID_cid = {'attr': {}} logger.debug(" Generating data...") - # dep_files = [os_path.join(input_dir, 'metanetx', f) for f in rrCache.__cache['cid_strc']['deps']['file_deps']] - dep_files = [os_path.join(input_dir, f) for f in rrCache.__cache['cid_strc']['deps']['file_deps']] - dep_files += [os_path.join(input_dir, f) for f in rrCache.__cache['cid_xref']['deps']['file_deps']] - cid_strc, cid_name = rrCache._m_mnxm_strc(dep_files, interactive=interactive, logger=logger) - - # if deprecatedCID_cid['attr'] != {}: - # # print(cid_strc['MNXM1106057']) - # # Replace compound IDs that have no structure with one that has. - # # Done from a manually built file - # with open(os_path.join(input_dir, 'MNXM_replacement_20190524.csv')) as csv_file: - # reader = csv_reader(csv_file, delimiter=' ') - # for row in reader: - # if not row[0].startswith('#') and len(row) > 1: - # if row[1] != 'R_group': - # # print(row) - # cid_strc[row[0]] = cid_strc[row[1]] + dep_files = {} + # Iterate over the file dependencies and find the corresponding files in the input cache sources + for dep_file in rrCache.__cache["cid_strc"]["deps"]["file_deps"]: + # Iterate over sources to look if the dependency file is listed in + for scat, source in rrCache.__cache_sources.items(): + if scat not in dep_files: + dep_files[scat] = [] + if dep_file in source["files"]: + # Look is the file is listed for one of the databases to use, + # i.e. if the value is the fingerprint or a dict + if isinstance(source["files"][dep_file], dict): + for db in source["files"][dep_file]: + if db in databases: + dep_files[scat].append( + os_path.join(input_dir, db, dep_file) + ) + else: + dep_files[scat].append(os_path.join(input_dir, dep_file)) + 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) - # if cid_name: - # rrCache._store_cache_to_file(cid_name, f_cid_name, logger=logger) - # else: - # logger.debug(" No cid_name file found in cache, skipping generation of cid_name") - - 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')) + attribute = "inchikey_cid" + logger.debug(c_attr("bold") + attribute + c_attr("reset")) inchikey_cid = None - f_inchikey_cid = os_path.join(outdir, rrCache.__cache[attribute]['file']['name']) + f_inchikey_cid = os_path.join( + outdir, rrCache.__cache[attribute]["file"]["name"] + ) # Do not checksum since it is a dictionary if os_path.exists(f_inchikey_cid) and check_sha( - f_inchikey_cid, - rrCache.__cache[attribute] + f_inchikey_cid, rrCache.__cache[attribute] ): logger.debug(" Cache file already exists") else: - if not cid_strc['attr']: + if not cid_strc["attr"]: logger.debug(" Loading input data from file...") - cid_strc['attr'] = rrCache._load_json(cid_strc['file']) + cid_strc["attr"] = rrCache._load_json(cid_strc["file"]) logger.debug(" Generating data...") - inchikey_cid = rrCache._m_inchikey_cid(cid_strc['attr']) + inchikey_cid = rrCache._m_inchikey_cid(cid_strc["attr"]) logger.debug(" Writing data to file...") rrCache._store_cache_to_file(inchikey_cid, f_inchikey_cid, logger=logger) del inchikey_cid - @staticmethod def _gen_cid_xref( input_dir: str, outdir: str, - # deprecatedCID_cid: Dict, - logger: Logger = getLogger(__name__) - ) -> Dict: - attribute = 'cid_xref' - logger.debug(c_attr('bold')+attribute+c_attr('reset')) + logger: Logger = LOGGER, + ) -> dict: + logger.debug(f"input_dir: {input_dir}") + logger.debug(f"outdir: {outdir}") + + attribute = "cid_xref" + logger.debug(c_attr("bold") + attribute + c_attr("reset")) cid_xref = None - f_cid_xref = os_path.join(outdir, rrCache.__cache[attribute]['file']['name']) + f_cid_xref = os_path.join(outdir, rrCache.__cache[attribute]["file"]["name"]) # Do not checksum since it is a dictionary if os_path.exists(f_cid_xref) and check_sha( - f_cid_xref, - rrCache.__cache[attribute] + f_cid_xref, rrCache.__cache[attribute] ): cid_xref = rrCache._load_json(f_cid_xref) logger.debug(" Cache file already exists") else: - # if not deprecatedCID_cid['attr']: - # logger.debug(" Loading input data from file...") - # deprecatedCID_cid['attr'] = rrCache._load_json(deprecatedCID_cid['file']) logger.debug(" Generating data...") - cid_xref = rrCache._m_mnxm_xref( - os_path.join(input_dir, 'chem_xref.tsv'), - # deprecatedCID_cid['attr'] - ) + dep_files = [ + os_path.join(input_dir, f) + for f in rrCache.__cache[attribute]["deps"]["file_deps"] + ] + logger.debug(f" dep_files: {dep_files}") + cid_xref = rrCache._m_mnxm_xref(dep_files[0]) logger.debug(" Writing data to file...") rrCache._store_cache_to_file(cid_xref, f_cid_xref, logger=logger) - return { - 'attr': cid_xref, - 'file': f_cid_xref - } - - - @staticmethod - def _gen_chebi_cid( - input_dir: str, - outdir: str, - cid_xref: Dict, - logger: Logger = getLogger(__name__) - ) -> Dict: - attribute = 'chebi_cid' - logger.debug(c_attr('bold')+attribute+c_attr('reset')) - chebi_cid = None - f_chebi_cid = os_path.join(outdir, rrCache.__cache[attribute]['file']['name']) - - # Do not checksum since it is a dictionary - if os_path.exists(f_chebi_cid) and check_sha( - f_chebi_cid, - rrCache.__cache[attribute] - ): - logger.debug(" Cache file already exists") - else: - logger.debug(" Generating data...") - chebi_cid = rrCache._m_chebi_cid(cid_xref['attr']) - # print_OK() - logger.debug(" Writing data to file...") - rrCache._store_cache_to_file(chebi_cid, f_chebi_cid, logger=logger) - del chebi_cid - # print_OK() - + return {"attr": cid_xref, "file": f_cid_xref} + + # @staticmethod + # def _gen_chebi_cid( + # input_dir: str, + # outdir: str, + # cid_xref: Dict, + # logger: Logger = LOGGER, + # ) -> Dict: + # attribute = "chebi_cid" + # logger.debug(c_attr("bold") + attribute + c_attr("reset")) + # chebi_cid = None + # f_chebi_cid = os_path.join(outdir, rrCache.__cache[attribute]["file"]["name"]) + + # # Do not checksum since it is a dictionary + # if os_path.exists(f_chebi_cid) and check_sha( + # f_chebi_cid, rrCache.__cache[attribute] + # ): + # logger.debug(" Cache file already exists") + # else: + # logger.debug(" Generating data...") + # chebi_cid = rrCache._m_chebi_cid(cid_xref["attr"]) + # # print_OK() + # logger.debug(" Writing data to file...") + # rrCache._store_cache_to_file(chebi_cid, f_chebi_cid, logger=logger) + # del chebi_cid + # # print_OK() @staticmethod def _gen_deprecatedRID_rid( - input_dir: str, - outdir: str, - logger: Logger = getLogger(__name__) - ) -> Dict: - attribute = 'deprecatedRID_rid' - logger.debug(c_attr('bold')+attribute+c_attr('reset')) + input_dir: str, outdir: str, logger: Logger = LOGGER + ) -> dict: + attribute = "deprecatedRID_rid" + logger.debug(c_attr("bold") + attribute + c_attr("reset")) deprecatedRID_rid = None - f_deprecatedRID_rid = os_path.join(outdir, rrCache.__cache[attribute]['file']['name']) + f_deprecatedRID_rid = os_path.join( + outdir, rrCache.__cache[attribute]["file"]["name"] + ) # Do not checksum since it is a dictionary if os_path.exists(f_deprecatedRID_rid) and check_sha( - f_deprecatedRID_rid, - rrCache.__cache[attribute] + f_deprecatedRID_rid, rrCache.__cache[attribute] ): deprecatedRID_rid = rrCache._load_json(f_deprecatedRID_rid) logger.debug(" Cache file already exists") else: logger.debug(" Generating data...") deprecatedRID_rid = rrCache._m_deprecatedMNXR( - os_path.join(input_dir, 'reac_xref.tsv') + os_path.join(input_dir, "reac_xref.tsv") ) logger.debug(" Writing data to file...") - rrCache._store_cache_to_file(deprecatedRID_rid, f_deprecatedRID_rid, logger=logger) - - return { - 'attr': deprecatedRID_rid, - 'file': f_deprecatedRID_rid - } + rrCache._store_cache_to_file( + deprecatedRID_rid, f_deprecatedRID_rid, logger=logger + ) + return {"attr": deprecatedRID_rid, "file": f_deprecatedRID_rid} @staticmethod - def _gen_rr_reactions( + def _gen_reactions( input_dir: str, outdir: str, - type: str = 'legacy', - logger: Logger = getLogger(__name__) + type: str = "legacy", + databases: list[str] = DEFAULTS["databases"], + attribute: str | None = None, + logger: Logger = LOGGER, ) -> None: - attribute = 'rr_reactions' - logger.debug(c_attr('bold')+attribute+c_attr('reset')) - rr_reactions = None - f_rr_reactions = os_path.join(outdir, rrCache.__cache[attribute]['file']['name']) + logger.debug(f"input_dir: {input_dir}") + logger.debug(f"outdir: {outdir}") + logger.debug(f"type: {type}") + logger.debug(f"databases: {databases}") + + if attribute is None: + logger.error("Attribute must be specified for reaction generation") + return + _attribute = f"{attribute}_reactions" + logger.debug(c_attr("bold") + attribute + c_attr("reset")) + reactions = {} + + outfile = os_path.join(outdir, rrCache.__cache[_attribute]["file"]["name"]) # Do not checksum since it is a dictionary - if os_path.exists(f_rr_reactions) and check_sha( - f_rr_reactions, - rrCache.__cache[attribute] - ): + if os_path.exists(outfile) and check_sha(outfile, rrCache.__cache[_attribute]): logger.debug(" Cache file already exists") else: - logger.debug(" Generating data...") - # dep_files = [os_path.join(input_dir, 'metanetx', f) for f in rrCache.__cache[attribute]['deps']['file_deps']] - dep_files = [os_path.join(input_dir, f) for f in rrCache.__cache[attribute]['deps']['file_deps']] - if type == 'legacy': - rr_reactions = rrCache._m_rr_reactions_legacy( - dep_files[0], - logger=logger - ) - else: - rr_reactions = rrCache._m_rr_reactions( - dep_files[0], - logger=logger - ) + dep_files = {} + # Iterate over the file dependencies and find the corresponding files in the input cache sources + for dep_file in rrCache.__cache[_attribute]["deps"]["file_deps"]: + # Iterate over sources to look if the dependency file is listed in + for scat, source in rrCache.__cache_sources.items(): + if scat not in dep_files: + dep_files[scat] = [] + if dep_file in source["files"]: + # Look is the file is listed for one of the databases to use, + # i.e. if the value is the fingerprint or a dict + if isinstance(source["files"][dep_file], dict): + for db in source["files"][dep_file]: + if db in databases: + dep_files[scat].append( + os_path.join(input_dir, db, dep_file) + ) + else: + logger.debug( + f"Database {db} is not in the list of databases to include in the cache, skipping generation of reactions for {db}" + ) + continue + else: + dep_files[scat].append(os_path.join(input_dir, dep_file)) + + method_name = "_m_" + attribute + "_reactions" + + if type == "legacy": + method_name += "_legacy" + # Get 'rr2' if not empty otherwise 'rr2more' + if dep_files["rr2"] != []: + dep_files = dep_files["rr2"][0] + elif dep_files["rr2more"] != []: + dep_files = dep_files["rr2more"][0] + else: + logger.error( + "No reaction rule file found for legacy type, cannot generate reactions" + ) + return + reactions = getattr(rrCache, method_name)(dep_files, logger=logger) logger.debug(" Writing data to file...") + rrCache._store_cache_to_file(reactions, outfile, logger=logger) - rrCache._store_cache_to_file(rr_reactions, f_rr_reactions, logger=logger) - - del rr_reactions - + del 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')) + attribute = "comp_xref, deprecatedCompID_compid" + logger.debug(c_attr("bold") + attribute + c_attr("reset")) comp_xref = deprecatedCompID_compid = None - f_comp_xref = os_path.join(outdir, rrCache.__cache['comp_xref']['file']['name']) - f_deprecatedCompID_compid = os_path.join(outdir, rrCache.__cache['deprecatedCompID_compid']['file']['name']) + f_comp_xref = os_path.join(outdir, rrCache.__cache["comp_xref"]["file"]["name"]) + f_deprecatedCompID_compid = os_path.join( + outdir, rrCache.__cache["deprecatedCompID_compid"]["file"]["name"] + ) # Do not checksum since it is a dictionary - if os_path.exists(f_comp_xref) and check_sha( - f_comp_xref, - rrCache.__cache['comp_xref'] - ) and os_path.exists(f_deprecatedCompID_compid) and check_sha( - f_deprecatedCompID_compid, - rrCache.__cache['deprecatedCompID_compid'] + if ( + os_path.exists(f_comp_xref) + and check_sha(f_comp_xref, rrCache.__cache["comp_xref"]) + and os_path.exists(f_deprecatedCompID_compid) + and check_sha( + f_deprecatedCompID_compid, rrCache.__cache["deprecatedCompID_compid"] + ) ): logger.debug(" Cache files already exist") # print_OK() else: logger.debug(" Generating data...") comp_xref, deprecatedCompID_compid = rrCache._m_mnxc_xref( - os_path.join(input_dir, 'comp_xref.tsv') + os_path.join(input_dir, "comp_xref.tsv") ) # print_OK() logger.debug(" Writing data to file...") @@ -779,161 +922,86 @@ def _gen_comp_xref_deprecatedCompID_compid( # print_OK() del comp_xref logger.debug(" Writing data to file...") - rrCache._store_cache_to_file(deprecatedCompID_compid, f_deprecatedCompID_compid, logger=logger) + rrCache._store_cache_to_file( + deprecatedCompID_compid, f_deprecatedCompID_compid, logger=logger + ) # print_OK() del deprecatedCompID_compid - - @staticmethod - def _gen_template_reactions( - input_dir: str, - outdir: str, - # deprecatedRID_rid: Dict, - type: str = 'legacy', - logger: Logger = getLogger(__name__) - ) -> None: - logger.debug('Generating template_reactions') - logger.debug('input_dir: '+str(input_dir)) - logger.debug('outdir: '+str(outdir)) - attribute = 'template_reactions' - logger.debug(c_attr('bold')+attribute+c_attr('reset')) - template_reactions = None - f_template_reactions = os_path.join(outdir, rrCache.__cache[attribute]['file']['name']) - - # if os_path.exists(f_template_reactions) and check_sha( - # f_template_reactions, - # rrCache.__cache[attribute] - # ): - # logger.debug(" Cache file already exists") - # else: - logger.debug(" Generating data...") - - # dep_files = [os_path.join(input_dir, 'metanetx', f) for f in rrCache.__cache[attribute]['deps']['file_deps']] - dep_files = [os_path.join(input_dir, f) for f in rrCache.__cache[attribute]['deps']['file_deps']] - if type == 'legacy': - template_reactions = rrCache._m_template_reactions_legacy(dep_files[0], logger=logger) - # if deprecatedRID_rid: - # # Handle deprecated reaction IDs - # for depRID, newRID in deprecatedRID_rid['attr'].items(): - # try: - # template_reactions[depRID] = template_reactions[newRID] - # except KeyError as key: - # logger.warning(f'Reaction ID {key} not found in {dep_files[0]}') - else: - template_reactions = rrCache._m_template_reactions(dep_files[0], logger=logger) - - logger.debug(" Writing data to file...") - - rrCache._store_cache_to_file(template_reactions, f_template_reactions, logger=logger) - - del template_reactions - - - # def _load_from_file(self, filename): - # self.logger.debug("Loading "+filename+"...") - # data = self._load_json( - # os_path.join(self.__cache_dir, filename) - # ) - # return data - - def _check_or_load_cache(self): - self.logger.debug('Checking cache...') - print_start(self.logger, 'Loading cache in memory') + self.logger.debug("Checking cache...") + print_start(self.logger, "Loading cache in memory") for attribute in self.__attributes_list: - filename = rrCache.__cache[attribute]['file']['name'] - self.logger.debug('Loading '+attribute+' from '+filename+'...') + filename = rrCache.__cache[attribute]["file"]["name"] + self.logger.debug("Loading " + attribute + " from " + filename + "...") if self.get(attribute) is None: self.set( - attribute, - self._load_json( - os_path.join(self.__cache_dir, filename) - ) + attribute, self._load_json(os_path.join(self.__cache_dir, filename)) ) - dico = self._load_json( - os_path.join(self.__cache_dir, filename) - ) print_progress(self.logger) else: - self.logger.debug(attribute+" already loaded in memory") + self.logger.debug(attribute + " already loaded in memory") print_end(self.logger) - @staticmethod 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): makedirs(outdir, exist_ok=True) filename = os_path.join(outdir, file) - logger.debug(f'Checking {file}...') + logger.debug(f"Checking {file}...") # check if file exists if os_path.isfile(filename): - logger.debug(f'File {file} present') + logger.debug(f"File {file} present") else: - logger.debug(f'File {filename} not present') + logger.debug(f"File {filename} not present") # start_time = time_time() rrCache.__download_input_cache(url, file, outdir) - # # if file size exceeds 20MB, compress it (compatible with all OS) - # if os_path.getsize(filename) > 20 * 1024 * 1024: - # logger.debug(f'File {file} size exceeds 20MB, compressing...') - # with NamedTemporaryFile(delete=False) as temp_file: - # temp_file.close() - # # Compress the file - # with gzip_open(temp_file.name, 'wb') as f_out: - # with open(filename, 'rb') as f_in: - # f_out.writelines(f_in) - # # Replace the original file with the compressed one - # os_rename(temp_file.name, filename+'.gz') - # logger.debug(f'File {file} compressed successfully') print_progress(logger) # end_time = time_time() # check sha512sum if check_sha(filename, fingerprint): - logger.debug(f'File {file} present and sha ok') + logger.debug(f"File {file} present and sha ok") else: # sha not ok - logger.debug(f'\n\ + logger.debug(f"\n\ filename: {filename}\n\ sha (computed): {sha512(Path(filename).read_bytes()).hexdigest()}\n\ sha (expected): {fingerprint}\n\ - ' + ") + raise FileCorruptedError( + f"Unable to download input-cache file {file}. Either the URL is broken or the file content has changed." ) - raise FileCorruptedError(f'Unable to download input-cache file {file}. Either the URL is broken or the file content has changed.') - @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: url (str): The URL to download the file from. file (str): The name of the file to download. outdir (str): The directory to save the downloaded file. """ - logger.debug(f'Downloading {file} from {url}') + logger.debug(f"Downloading {file} from {url}") # Check if the URL is empty if not url: - logger.error(f'\n\n*** URL is empty, cannot download file {file}.\n') + logger.error(f"\n\n*** URL is empty, cannot download file {file}.\n") # Check if the output directory exists, if not create it if not os_path.isdir(outdir): - makedirs(outdir ,exist_ok=True) + makedirs(outdir, exist_ok=True) - logger.debug(f'Downloading {file} from {url}') - download(url+file, os_path.join(outdir, file)) + logger.debug(f"Downloading {file} from {url}") + download(url + file, os_path.join(outdir, file)) ########################################################## ################## Private Functions ##################### @@ -947,15 +1015,15 @@ def __download_input_cache( # @param filename File to fetch data from # @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 # @@ -965,22 +1033,26 @@ 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 - json_bytes = json_dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8") + json_bytes = json_dumps(data, sort_keys=True, separators=(",", ":")).encode( + "utf-8" + ) # 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)) @@ -1010,7 +1082,6 @@ def _checkRIDdeprecated(rid, deprecatedRID_rid): ################## Public functions ############################# ################################################################# - ########################### MNX parsers ############################# ## Function to parse the chem_xref.tsv and reac_xref.tsv file of MetanetX @@ -1022,16 +1093,15 @@ def _checkRIDdeprecated(rid, deprecatedRID_rid): # @return Dictionnary of identifiers # TODO: save the self.deprecatedCID_cid to be used in case there rp_paths uses an old version of MNX - @staticmethod def _deprecatedMNX(xref_path): deprecatedMNX_mnx = {} - with open(xref_path, 'rt') as f: - c = csv_reader(f, delimiter='\t') + with open(xref_path, "rt") as f: + c = csv_reader(f, delimiter="\t") for row in c: - if not row[0][0] == '#': - mnx = row[0].split(':') - if mnx[0] == 'deprecated': + if row[0][0] != "#": + mnx = row[0].split(":") + if mnx[0] == "deprecated": deprecatedMNX_mnx[mnx[1]] = row[1] return deprecatedMNX_mnx @@ -1044,7 +1114,7 @@ def _m_deprecatedMNXM(chem_xref_path): deprecatedCID_cid = {} deprecatedCID_cid = rrCache._deprecatedMNX(chem_xref_path) deprecatedCID_cid.update(rrCache.__convertMNXM) - deprecatedCID_cid['MNXM01'] = 'MNXM1' + deprecatedCID_cid["MNXM01"] = "MNXM1" return deprecatedCID_cid ## Function to parse the reac_xref.tsv file of MetanetX @@ -1059,7 +1129,6 @@ def _m_deprecatedMNXM(chem_xref_path): def _m_deprecatedMNXR(reac_xref_path): return rrCache._deprecatedMNX(reac_xref_path) - ## Function to parse the chemp_prop.tsv file from MetanetX and compounds.tsv from RetroRules. Uses the InchIkey as key to the dictionnary # # Generate a dictionnary gaving the formula, smiles, inchi and inchikey for the components @@ -1072,15 +1141,14 @@ def _m_deprecatedMNXR(reac_xref_path): # @return cid_strc Dictionnary of formula, smiles, inchi and inchikey @staticmethod def _m_mnxm_strc( - paths: List[str], + paths: dict[str, list[str]], # deprecatedCID_cid: Dict = None, - interactive: bool = DEFAULTS['interactive'], - logger: Logger = getLogger(__name__) - ) -> Tuple[Dict, Dict]: - + interactive: bool = DEFAULTS["interactive"], + 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 (List[str]): List of paths to the input files (rr_compounds_path first, then chem_prop_path). + paths (Dict[str, List[str]]): Dictionary mapping file types to lists of paths. deprecatedCID_cid (Dict): Dictionary of deprecated CID to cid. interactive (bool): Whether to ask the user for confirmation before overwriting existing files. Returns: @@ -1089,60 +1157,69 @@ def _m_mnxm_strc( - cid_name: Dictionary of compound names. """ - logger.debug(f'paths: {paths}') + logger.debug(f"paths: {paths}") # logger.debug(f'deprecatedCID_cid: {deprecatedCID_cid}') - logger.debug(f'interactive: {interactive}') + logger.debug(f"interactive: {interactive}") - rr_compounds_path = paths[0] - chem_prop_path = paths[1] if len(paths) > 1 else "" - comp_xref_path = paths[2] if len(paths) > 2 else "" + rr_compounds_paths = paths["rr2more"] + chem_prop_path = ( + paths["mnx"][0] if "mnx" in paths and len(paths["mnx"]) > 0 else None + ) + comp_xref_path = ( + paths["mnx"][1] if "mnx" in paths and len(paths["mnx"]) > 1 else None + ) + + logger.debug(f"rr_compounds_paths: {rr_compounds_paths}") + logger.debug(f"chem_prop_path: {chem_prop_path}") + logger.debug(f"comp_xref_path: {comp_xref_path}") cid_strc = {} cid_name = {} - # 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"]}') - # for key in tmp.keys(): - # try: - # tmp[key] = row[key] - # except KeyError: - # logger.debug(f'No {key} in RetroRules {rr_compounds_path} for '+str(row['cid'])+', setting to None') - - cid_strc[tmp['cid']] = tmp + for rr_compounds_path in rr_compounds_paths: + # Parse the compounds.tsv file from RetroRules + 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 if chem_prop_path: # Parse the chem_prop.tsv file from MetanetX - with open(chem_prop_path, 'rt') as f: + with open(chem_prop_path, "rt", encoding="utf-8-sig") as f: tmp = {} # read CSV with both tab and space as delimiters - c = csv_reader(f, delimiter='\t') + c = csv_reader(f, delimiter="\t") for row in c: - if row[0].startswith('#'): + if row[0].startswith("#"): header = row else: # Set 'tmp' in a generic way for i in range(len(header)): # remove '#' from column field and # convert to lower case - field = header[i].replace('#', '').lower() - if field == 'id': - field = 'cid' + field = header[i].replace("#", "").lower() + if field == "id": + field = "cid" tmp[field] = row[i] if interactive: print() @@ -1152,7 +1229,7 @@ def _m_mnxm_strc( # mnxm = rrCache._checkCIDdeprecated(row[0], deprecatedCID_cid) mnxm = row[0] if interactive: - print(f'Converted into {mnxm}') + print(f"Converted into {mnxm}") # tmp = { # 'formula': row[2], # 'smiles': row[6], @@ -1162,24 +1239,28 @@ def _m_mnxm_strc( # 'name': row[1] # } for i in tmp: - if tmp[i] == '' or tmp[i] == 'NA': + if tmp[i] == "" or tmp[i] == "NA": tmp[i] = None - try: # in MetaNetX 3.1, there is no name field - if mnxm not in cid_name and tmp['name']: - cid_name[mnxm] = tmp['name'] + try: # in MetaNetX 3.1, there is no name field + if mnxm not in cid_name and tmp["name"]: + cid_name[mnxm] = tmp["name"] except KeyError: # If the name is not present - logger.debug('No name in chem_prop.tsv for '+str(mnxm)+', setting to None') + logger.debug( + "No name in chem_prop.tsv for " + + str(mnxm) + + ", setting to None" + ) # Compound already in the dictionnary if mnxm in cid_strc: # # If the ID has been converted, then create a link # if mnxm != row[0]: # cid_strc[mnxm] = cid_strc[row[0]] if interactive: - print('already in cid_strc') - cid_strc[mnxm]['formula'] = row[2] - cid_strc[mnxm]['name'] = row[1] - for key in ['inchi', 'inchikey', 'smiles']: + print("already in cid_strc") + cid_strc[mnxm]["formula"] = row[2] + cid_strc[mnxm]["name"] = row[1] + for key in ["inchi", "inchikey", "smiles"]: if not cid_strc[mnxm][key] and tmp[key]: cid_strc[mnxm][key] = tmp[key] # else: # Compound not in the dictionnary @@ -1224,24 +1305,24 @@ def _m_mnxm_strc( print(cid_strc[mnxm]) interactive = ask_user_input() else: - logger.debug('No chem_prop.tsv or deprecatedCID_cid provided, skipping structure parsing') + logger.debug( + "No chem_prop.tsv or deprecatedCID_cid provided, skipping structure parsing" + ) if comp_xref_path: # Add cross references from comp_xref.tsv - comp_xref, _ = rrCache._m_mnxc_xref( - comp_xref_path, - logger=logger - ) + comp_xref, _ = rrCache._m_mnxc_xref(comp_xref_path, logger=logger) for mnxc in comp_xref: if mnxc in cid_strc: - cid_strc[mnxc]['xref'] = comp_xref[mnxc] - logger.debug(f'Added cross references for {mnxc}: {comp_xref[mnxc]}') + cid_strc[mnxc]["xref"] = comp_xref[mnxc] + logger.debug( + f"Added cross references for {mnxc}: {comp_xref[mnxc]}" + ) else: - logger.debug('No comp_xref.tsv provided, skipping cross-reference parsing') + logger.debug("No comp_xref.tsv provided, skipping cross-reference parsing") # logger.removeHandler(logger.handlers[-1]) return cid_strc, cid_name - ## Function to parse the chem_xref.tsv file of MetanetX # # Generate a dictionnary of all cross references for a given chemical id (MNX) to other database id's @@ -1251,28 +1332,27 @@ def _m_mnxm_strc( # @return Dictionnary of cross references of a given chemical id # TODO: save the self.deprecatedCID_cid to be used in case there rp_paths uses an old version of MNX - @staticmethod 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') as f: - c = csv_reader(f, delimiter='\t') + 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: - dbName = 'mnx' + if len(row[0].split(":")) == 1: + dbName = "mnx" dbId = row[0] else: - dbName = row[0].split(':')[0] - dbId = ''.join(row[0].split(':')[1:]) - if dbName == 'deprecated': - dbName = 'mnx' + dbName = row[0].split(":")[0] + dbId = "".join(row[0].split(":")[1:]) + if dbName == "deprecated": + dbName = "mnx" # mnx if mnx not in cid_xref: cid_xref[mnx] = {} @@ -1287,7 +1367,6 @@ def _m_mnxm_xref( cid_xref[dbName][dbId] = mnx return cid_xref - ## Function to parse the comp_xref.tsv file of MetanetX # # Generate a dictionnary of compartments id's (MNX) to other database id's @@ -1296,36 +1375,32 @@ def _m_mnxm_xref( # @return a The dictionnary of compartment identifiers # 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') as f: - c = csv_reader(f, delimiter='\t') + with open(comp_xref_path, "rt", encoding="utf-8-sig") as f: + c = csv_reader(f, delimiter="\t") # 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: - dbName = 'mnx' + if len(row[0].split(":")) == 1: + dbName = "mnx" dbCompId = row[0] else: - dbName = row[0].split(':')[0] - dbCompId = ''.join(row[0].split(':')[1:]) + dbName = row[0].split(":")[0] + dbCompId = "".join(row[0].split(":")[1:]) dbCompId = dbCompId.lower() - if dbName == 'deprecated': - dbName = 'mnx' + if dbName == "deprecated": + dbName = "mnx" # create the dicts if mnxc not in comp_xref: comp_xref[mnxc] = {} @@ -1339,10 +1414,8 @@ def _m_mnxc_xref( return comp_xref, deprecatedCompID_compid - ######################## RetroRules specific functions ################## - ## Function to parse the rules_rall.tsv from RetroRules # # Extract from the reactions rules the ruleID, the reactionID, the direction of the rule directed to the origin reaction @@ -1354,86 +1427,132 @@ def _m_mnxc_xref( # @return Dictionnary describing each reaction rule @staticmethod - def _m_rr_reactions( - rules_rall_path: str, - logger: Logger = getLogger(__name__) - ) -> Dict: - rr_reactions = {} + def _m_rr_reactions(rules_rall_paths: str, logger: Logger = LOGGER) -> dict: + logger.debug(f"Parsing rules from {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 + _rules_rall_paths = rules_rall_paths["rr2"] - 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': 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)') - rr_reactions[row['TEMPLATE_ID']][row['REACTION_ID']]['rel_direction'] = 0 # bidirectional + rr_reactions = {} - return 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 + 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)" + ) + 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)+')') + 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 - - 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': 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 - } + for cid in row["Product_IDs"].split("."): + products[cid] = products.get(cid, 0) + 1 - 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 + 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, + } + + 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 ## Generate complete reactions from the rxn_recipes.tsv from RetroRules # @@ -1447,131 +1566,202 @@ def _m_rr_reactions_legacy( @staticmethod def _m_template_reactions( - metadata_path: str, - logger: Logger = getLogger(__name__) - ) -> Dict: + paths: dict[str, list[str]], logger: Logger = LOGGER + ) -> dict: - if not os_path.exists(metadata_path): - logger.error('Cannot find file: '+str(metadata_path)) - return None + metadata_paths = paths["rr2"] + reac_prop_path = ( + paths["mnx"][0] if "mnx" in paths and len(paths["mnx"]) > 0 else None + ) + + logger.debug(f"metadata_paths: {metadata_paths}") + logger.debug(f"reac_prop_path: {reac_prop_path}") reactions = {} - for row in csv_DictReader(gzip_open(metadata_path, 'rt'), delimiter='\t'): - if row['REACTION_ID'] not in reactions: - # print(row) - 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 []))) - # if row['REACTION_ID'] == 'MNXR182203': - # print(substrates) - # print(products) - # exit() - 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 + # 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 + 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 [] + ) + ) + ) + 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' + # Example of mnx_equation: 1 MNXM10958@MNXD1 + 1 MNXM1104529@MNXD1 = 1 MNXM1102128@MNXD1 + 1 MNXM8415@MNXD1 + if reac_prop_path: + reac_prop_df = rrCache.__load_reactions_tsv(reac_prop_path) + for _, row in reac_prop_df.iterrows(): + reac_id = row["ID"] + if reac_id not in reactions: + rxn = rrCache._read_equation(row["mnx_equation"], reac_id, logger) + # check 'right' and 'left' are not empty + if reac_id not in reactions and rxn["left"] and rxn["right"]: + reactions[reac_id] = { + "left": rxn["left"], + "right": rxn["right"], + "direction": 0, # default to bidirectional if not specified + "main_left": "", + "main_right": "", + } 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 + logger.debug( + f"Reaction {reac_id} already in reactions, skipping equation parsing" + ) return reactions + @staticmethod + def __load_reactions_tsv(path: str, logger: Logger = LOGGER) -> "DataFrame": + """ + Load a TSV file while: + - ignoring comment lines starting with '#' + - using the LAST commented line as the header + """ + + header = None + data_lines = [] + + with open(path, "r") as f: + for line in f: + line = line.rstrip("\n") + + if line.startswith("#"): + # Save last commented line as header + header = line[1:].split("\t") + else: + data_lines.append(line) + + if header is None: + raise ValueError("No header line found starting with '#'") + + # Rebuild TSV content without comments + tsv_content = "\n".join(data_lines) + + # Read with pandas + df = pd_read_csv(StringIO(tsv_content), sep="\t", names=header) + + return df @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)) + 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 + + # Direction + dir = rrCache._read_direction(row["Direction"], logger) + if dir is None: + # Pass to the next equation + continue + else: + rxn["direction"] = dir - 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 - - # Others - rxn['main_left'] = row['Main_left'].split(',') - rxn['main_right'] = row['Main_right'].split(',') - - reactions[row['#Reaction_ID']] = rxn + # Others + rxn["main_left"] = row["Main_left"].split(",") + rxn["main_right"] = row["Main_right"].split(",") - return reactions + reactions[row["#Reaction_ID"]] = rxn + 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 = LOGGER) -> dict: - def _read_equation( - eq: str, - rxn_id: str, - logger: Logger = getLogger(__name__) - ) -> Dict: - - if not 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)) + if len(eq.split("=")) != 2: + logger.warning( + "There should never be more or less than a left and right of an equation" + ) + logger.warning(f"Ignoring {eq}") return None #### for character matching that are returned DEFAULT_STOICHIO_RESCUE = { - '4n': 4, '3n': 3, '2n': 2, 'n': 1, - '(n)': 1, '(N)': 1, '(2n)': 2, '(x)': 1, - 'N': 1, 'm': 1, 'q': 1, - '0.01': 1, '0.1': 1, '0.5': 1, '1.5': 1, - '0.02': 1, '0.2': 1, - '(n-1)': 0, '(n-2)': -1 + "4n": 4, + "3n": 3, + "2n": 2, + "n": 1, + "(n)": 1, + "(N)": 1, + "(2n)": 2, + "(x)": 1, + "N": 1, + "m": 1, + "q": 1, + "0.01": 1, + "0.1": 1, + "0.5": 1, + "1.5": 1, + "0.02": 1, + "0.2": 1, + "(n-1)": 0, + "(n-2)": -1, } rxn = {} @@ -1580,11 +1770,11 @@ def _read_equation( # 0 = left, 1 = right for side in [0, 1]: rxn[side] = {} - if '@' in eq: - regex_str = r'(\(n-1\)|\d+|4n|3n|2n|n|\(n\)|\(N\)|\(2n\)|\(x\)|N|m|q|\(n\-2\)|\d+\.\d+) ([\w\d]+)@\w+' + if "@" in eq: + regex_str = r"(\(n-1\)|\d+|4n|3n|2n|n|\(n\)|\(N\)|\(2n\)|\(x\)|N|m|q|\(n\-2\)|\d+\.\d+) ([\w\d]+)@\w+" else: - regex_str = r'(\(n-1\)|\d+|4n|3n|2n|n|\(n\)|\(N\)|\(2n\)|\(x\)|N|m|q|\(n\-2\)|\d+\.\d+) ([\w\d]+)' - for spe in re_findall(regex_str, eq.split('=')[side]): + regex_str = r"(\(n-1\)|\d+|4n|3n|2n|n|\(n\)|\(N\)|\(2n\)|\(x\)|N|m|q|\(n\-2\)|\d+\.\d+) ([\w\d]+)" + for spe in re_findall(regex_str, eq.split("=")[side]): # 1) try to rescue if its one of the values try: # rxn[side][rrCache._checkCIDdeprecated(spe[1], deprecatedCID_cid)] = DEFAULT_STOICHIO_RESCUE[spe[0]] @@ -1598,17 +1788,13 @@ def _read_equation( ter = StreamHandler.terminator StreamHandler.terminator = "\n" logger.warning( - f'Cannot convert stoichio coeff {spe[0]} in {rxn_id}' + f"Cannot convert stoichio coeff {spe[0]} in {rxn_id}" ) StreamHandler.terminator = ter # Stop parsing this equation and pass the next return None - return { - 'left': rxn[0], - 'right': rxn[1] - } - + return {"left": rxn[0], "right": rxn[1]} # ------------ GENERIC FUNCTIONS ------------ # @@ -1628,16 +1814,20 @@ 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': + if itype == "smiles": return MolFromSmiles(idepic, sanitize=sanitize) - elif itype == 'inchi': + elif itype == "inchi": return MolFromInchi(idepic, sanitize=sanitize) else: raise NotImplementedError( f'"{itype}" is not a valid input type for MolFrom' ) + # Import (if needed) rdmol = MolFrom(idepic, itype, sanitize=True) if rdmol is None: # Check imprt @@ -1650,20 +1840,19 @@ 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': + if item == "smiles": # MolToSmiles is tricky, one mays want to check the possible options.. odepic[item] = MolToSmiles(rdmol) - elif item == 'inchi': + elif item == "inchi": odepic[item] = MolToInchi(rdmol) - elif item == 'inchikey': + elif item == "inchikey": odepic[item] = MolToInchiKey(rdmol) else: raise NotImplementedError(f'"{otype}" is not a valid output type') return odepic - # Function to parse the chem_xref.tsv file of MetanetX # # Generate a dictionnary of all cross references @@ -1679,8 +1868,8 @@ def MolFrom(idepic, itype, sanitize=True): def _m_chebi_cid(cid_xref): chebi_cid = {} for cid in cid_xref: - if 'chebi' in cid_xref[cid]: - for c in cid_xref[cid]['chebi']: + if "chebi" in cid_xref[cid]: + for c in cid_xref[cid]["chebi"]: chebi_cid[c] = cid return chebi_cid @@ -1693,11 +1882,11 @@ def _m_chebi_cid(cid_xref): def _m_inchikey_cid(cid_strc): inchikey_cid = {} for cid in cid_strc: - inchikey = cid_strc[cid]['inchikey'] + inchikey = cid_strc[cid]["inchikey"] # This line is needed to put a value in 'inchikey', # otherwise there are some problems in future strucutres if not inchikey: - inchikey = 'NO_INCHIKEY' + inchikey = "NO_INCHIKEY" if inchikey not in inchikey_cid: inchikey_cid[inchikey] = [] inchikey_cid[inchikey].append(cid) diff --git a/setup.py b/setup.py index d4acdc8..ffe87a7 100644 --- a/setup.py +++ b/setup.py @@ -1,63 +1,55 @@ -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' -url = 'https://github.com/brsynth/rr_cache' -authors = 'Joan Hérisson, Melchior du Lac' -corr_author = 'joan.herisson@univ-evry.fr' +package = "rr_cache" +descr = "Cache for RetroRules and MetaNetX" +url = "https://github.com/brsynth/rr_cache" +authors = "Joan Hérisson, Melchior du Lac" +corr_author = "joan.herisson@univ-evry.fr" ## LONG DESCRIPTION with open( - os_path.join( - os_path.dirname(os_path.realpath(__file__)), - 'README.md' - ), - 'r', - encoding='utf-8' + os_path.join(os_path.dirname(os_path.realpath(__file__)), "README.md"), + "r", + encoding="utf-8", ) as f: long_description = f.read() + def get_version(): with open( - os_path.join( - os_path.dirname(os_path.realpath(__file__)), - 'CHANGELOG.md' - ), - 'r' + os_path.join(os_path.dirname(os_path.realpath(__file__)), "CHANGELOG.md"), "r" ) as f: lines = f.readlines() for line in lines: - if line.startswith('##'): + if line.startswith("##"): from re import search - m = search("\[(.+)\]", line) + + m = search(r"\[(.+)\]", line) if m: return m.group(1) + setup( - name = package, - version = get_version(), - author = authors, - author_email = corr_author, - description = descr, - long_description = long_description, - long_description_content_type = 'text/markdown', - url = url, - packages = find_packages(), - package_dir = {package: package}, - include_package_data = True, - test_suite = 'pytest', - license = 'MIT', - classifiers = [ - 'Programming Language :: Python :: 3', - 'License :: OSI Approved :: MIT License', - 'Operating System :: OS Independent', + name=package, + version=get_version(), + author=authors, + author_email=corr_author, + description=descr, + long_description=long_description, + long_description_content_type="text/markdown", + url=url, + packages=find_packages(), + package_dir={package: package}, + include_package_data=True, + test_suite="pytest", + license="MIT", + classifiers=[ + "Programming Language :: Python :: 3", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", ], - python_requires = '>=3.7', + python_requires=">=3.7", ) - diff --git a/tests/data/compounds_mnx3.1.json b/tests/data/compounds_mnx3.1.json index b8f4f52..d3df8b7 100644 --- a/tests/data/compounds_mnx3.1.json +++ b/tests/data/compounds_mnx3.1.json @@ -1,10 +1,22 @@ { "MNXM2": { + "cid": "MNXM2", "formula": "H2O", - "smiles": "[H]O[H]", "inchi": "InChI=1S/H2O/h1H2", "inchikey": "XLYOFNOQVPJJNP-UHFFFAOYSA-N", - "cid": "MNXM2", - "name": "H2O" + "name": "H2O", + "smiles": "[H]O[H]", + "xref": { + "bigg": ["h2o", "m_h2o", "oh1", "m_oh1"], + "chebi": ["15377", "10743", "13352", "27313", "42043", "42857", "43228", "44292", "44701", "44819", "5585", "16234", "13365", "13419", "44641", "5594", "29412"], + "envipath": ["5882df9c-dae1-4d80-a40e-db4724271456/compound/969d0227-3069-4e44-9525-7ae7bad84170"], + "hmdb": ["hmdb01039", "hmdb02111"], + "kegg": ["c00001", "d00001", "c01328"], + "metacyc": ["oh", "oxonium", "water"], + "mnx": ["MNXM2", "mnxm114710", "mnxm11838", "mnxm527231", "mnxm89551"], + "reactome": ["109276", "113518", "113519", "113521", "141343", "1605715", "189422", "2022884", "29356", "351603", "5278291", "5668574", "5693747", "8851517"], + "sabiork": ["40"], + "seed": ["cpd00001", "cpd15275"] + } } } \ No newline at end of file diff --git a/tests/data/compounds_mnx4.4.json b/tests/data/compounds_mnx4.4.json index ffa17e9..4fd588b 100644 --- a/tests/data/compounds_mnx4.4.json +++ b/tests/data/compounds_mnx4.4.json @@ -1,3 +1,10 @@ { - "MNXM2": {"formula": null, "inchi": "InChI=1S/H2O/h1H2", "inchikey": "XLYOFNOQVPJJNP-UHFFFAOYSA-N", "cid": "MNXM2", "name": null, "smiles": null} + "MNXM2": { + "formula": null, + "inchi": "InChI=1S/H2O/h1H2", + "inchikey": "XLYOFNOQVPJJNP-UHFFFAOYSA-N", + "cid": "MNXM2", + "name": null, + "smiles": null + } } \ No newline at end of file diff --git a/tests/data/compounds_rr2026-v3.0.json b/tests/data/compounds_rr2026-v3.0.json new file mode 100644 index 0000000..e7d0456 --- /dev/null +++ b/tests/data/compounds_rr2026-v3.0.json @@ -0,0 +1,29 @@ +{ + "MNXM1": { + "cid": "MNXM1", + "formula": "mnx:PROTON", + "inchi": "InChI=1S/p+1", + "inchikey": "GPRLSGONYQIRFK-UHFFFAOYSA-N", + "name": "H(+)", + "smiles": "[H+]", + "xref": { + "CHEBI": ["15378", "24636", "29233", "29234"], + "bigg.metabolite": ["h"], "biggM": ["M_h", "h"], + "chebi": ["10744", "13357", "15378", "24636", "29233", "29234", "5584"], + "hmdb": ["HMDB0059597", "HMDB59597"], + "kegg.compound": ["C00080"], + "keggC": ["C00080", "M_C00080"], + "metacyc.compound": ["PROTON"], + "metacycM": ["PROTON"], + "mnx": ["MNXM1", "PROTON"], + "reactome": ["R-ALL-1132304", "R-ALL-113529", "R-ALL-1470067", "R-ALL-156540", "R-ALL-1614597", "R-ALL-163953", "R-ALL-193465", "R-ALL-194688", "R-ALL-2000349", "R-ALL-2429673", "R-ALL-2872447", "R-ALL-351626", "R-ALL-372511", "R-ALL-374900", "R-ALL-425969", "R-ALL-425978", "R-ALL-425999", "R-ALL-427899", "R-ALL-428040", "R-ALL-428548", "R-ALL-5228597", "R-ALL-5244410", "R-ALL-5668577", "R-ALL-6783901", "R-ALL-70106", "R-ALL-74722", "R-ALL-9631150", "R-ALL-9668967", "R-ALL-9683057"], + "reactomeM": ["R-ALL-1132304", "R-ALL-113529", "R-ALL-1470067", "R-ALL-156540", "R-ALL-1614597", "R-ALL-163953", "R-ALL-193465", "R-ALL-194688", "R-ALL-2000349", "R-ALL-2429673", "R-ALL-2872447", "R-ALL-351626", "R-ALL-372511", "R-ALL-374900", "R-ALL-425969", "R-ALL-425978", "R-ALL-425999", "R-ALL-427899", "R-ALL-428040", "R-ALL-428548", "R-ALL-5228597", "R-ALL-5244410", "R-ALL-5668577", "R-ALL-6783901", "R-ALL-70106", "R-ALL-74722", "R-ALL-9631150", "R-ALL-9668967", "R-ALL-9683057"], + "sabiork.compound": ["39"], + "sabiorkM": ["39"], + "seed.compound": ["cpd00067"], + "seedM": ["M_cpd00067", "cpd00067"], + "vmhM": ["HC02115", "M_HC02115", "M_h", "h"], + "vmhmetabolite": ["HC02115", "h"] + } + } +} \ No newline at end of file diff --git a/tests/data/metrics_mnx3.1.json b/tests/data/metrics_mnx3.1.json index 40af2e8..3f0a622 100644 --- a/tests/data/metrics_mnx3.1.json +++ b/tests/data/metrics_mnx3.1.json @@ -4,11 +4,11 @@ "file_size": 2786801 }, "cid_strc": { - "length": 694316, + "length": 468224, "file_size": 296896910 }, "cid_xref": { - "length": 694321, + "length": 694324, "file_size": 88383985 }, "comp_xref": { @@ -28,11 +28,11 @@ "file_size": 1437122 }, "inchikey_cid": { - "length": 332114, + "length": 332019, "file_size": 20071352 }, "template_reactions": { - "length": 93329, + "length": 44320, "file_size": 7643885 }, "rr_reactions": { diff --git a/tests/data/metrics_mnx4.4.json b/tests/data/metrics_mnx4.4.json index a9c4b32..0c9021a 100644 --- a/tests/data/metrics_mnx4.4.json +++ b/tests/data/metrics_mnx4.4.json @@ -1,40 +1,12 @@ { - "chebi_cid": { - "length": 165937, - "file_size": 2786801 - }, - "cid_name": { - "length": 1292152, - "file_size": 101757526 - }, "cid_strc": { - "length": 1314786, + "length": 468224, "file_size": 806612376 }, "cid_xref": { - "length": 1292180, + "length": 1292182, "file_size": 243759725 }, - "comp_xref": { - "length": 44, - "file_size": 3824 - }, - "deprecatedCID_cid": { - "length": 8, - "file_size": 179 - }, - "deprecatedCompID_compid": { - "length": 128, - "file_size": 30 - }, - "deprecatedRID_rid": { - "length": 0, - "file_size": 2 - }, - "inchikey_cid": { - "length": 937156, - "file_size": 55056834 - }, "template_reactions": { "length": 44320, "file_size": 8519350 diff --git a/tests/data/metrics_rr2026-v3.0.json b/tests/data/metrics_rr2026-v3.0.json new file mode 100644 index 0000000..1378aee --- /dev/null +++ b/tests/data/metrics_rr2026-v3.0.json @@ -0,0 +1,18 @@ +{ + "cid_strc": { + "length": 1237578, + "file_size": 806612376 + }, + "cid_xref": { + "length": 1495699, + "file_size": 243759725 + }, + "template_reactions": { + "length": 95419, + "file_size": 8519350 + }, + "rr_reactions": { + "length": 361751, + "file_size": 84656878 + } +} \ No newline at end of file diff --git a/tests/data/reactions.json b/tests/data/reactions_mnx3.1.json similarity index 100% rename from tests/data/reactions.json rename to tests/data/reactions_mnx3.1.json diff --git a/tests/data/reactions_mnx4.4.json b/tests/data/reactions_mnx4.4.json new file mode 100644 index 0000000..5b802fe --- /dev/null +++ b/tests/data/reactions_mnx4.4.json @@ -0,0 +1,19 @@ +{ + "MNXR94688": { + "left": { + "MNXM1": 1, + "MNXM17": 1, + "MNXM91349": 1 + }, + "right": { + "MNXM52": 1 + }, + "direction": 0, + "main_left": [ + "MNXM91349" + ], + "main_right": [ + "MNXM52" + ] + } +} \ No newline at end of file diff --git a/tests/data/reactions_rr2026-v3.0.json b/tests/data/reactions_rr2026-v3.0.json new file mode 100644 index 0000000..97d5528 --- /dev/null +++ b/tests/data/reactions_rr2026-v3.0.json @@ -0,0 +1,16 @@ +{ + "RHEA:84375": { + "direction": 0, + "left": { + "CHEBI:48945": 1, + "CHEBI:58349": 1 + }, + "main_left": "CHEBI:48945", + "main_right": "CHEBI:16182", + "right": { + "CHEBI:15378": 1, + "CHEBI:16182": 1, + "CHEBI:57783": 1 + } + } +} diff --git a/tests/data/retrorules.json b/tests/data/retrorules_mnx3.1.json similarity index 100% rename from tests/data/retrorules.json rename to tests/data/retrorules_mnx3.1.json diff --git a/tests/data/retrorules_mnx4.4.json b/tests/data/retrorules_mnx4.4.json new file mode 100644 index 0000000..5bb4fab --- /dev/null +++ b/tests/data/retrorules_mnx4.4.json @@ -0,0 +1,46 @@ +{ + "RR-02-f85f00f767901186-16-F": { + "MNXR100137": { + "rule_id": "RR-02-f85f00f767901186-16-F", + "rule_score": 0.7486085128675456, + "reac_id": "MNXR100137", + "subs_id": "MNXM2210", + "rel_direction": 1, + "left": { + "MNXM2210": 1 + }, + "right": { + "MNXM139": 1, + "MNXM83": 1 + } + }, + "MNXR112247": { + "rule_id": "RR-02-f85f00f767901186-16-F", + "rule_score": 0.7486085128675456, + "reac_id": "MNXR112247", + "subs_id": "MNXM7283", + "rel_direction": 1, + "left": { + "MNXM7283": 1 + }, + "right": { + "MNXM2473": 1, + "MNXM83": 1 + } + }, + "MNXR116653": { + "rule_id": "RR-02-f85f00f767901186-16-F", + "rule_score": 0.7486085128675456, + "reac_id": "MNXR116653", + "subs_id": "MNXM13314", + "rel_direction": 1, + "left": { + "MNXM13314": 1 + }, + "right": { + "MNXM13541": 1, + "MNXM83": 1 + } + } + } +} diff --git a/tests/data/retrorules_rr2026-v3.0.json b/tests/data/retrorules_rr2026-v3.0.json new file mode 100644 index 0000000..5b4ae97 --- /dev/null +++ b/tests/data/retrorules_rr2026-v3.0.json @@ -0,0 +1,88 @@ +{ + "RR:03-FFFFD8-869728-14F123": { + "MNXR112980": { + "left": { + "MNXM2376": 1 + }, + "left_excluded": [ + "MNXM8", + "MNXM2376" + ], + "reac_id": "MNXR112980", + "rel_direction": -1, + "right": { + "MNXM2526": 1 + }, + "right_excluded": [ + "MNXM1", + "MNXM10" + ], + "rule_id": "RR:03-FFFFD8-869728-14F123", + "rule_score": 0.8408964, + "subs_id": "MNXM2376" + }, + "MNXR112982": { + "left": { + "MNXM2376": 1 + }, + "left_excluded": [ + "MNXM2376", + "MNXM732620" + ], + "reac_id": "MNXR112982", + "rel_direction": 1, + "right": { + "MNXM2526": 1 + }, + "right_excluded": [ + "WATER", + "WATER" + ], + "rule_id": "RR:03-FFFFD8-869728-14F123", + "rule_score": 0.8408964, + "subs_id": "MNXM2376" + }, + "RHEA:27433": { + "left": { + "CHEBI:59895": 1 + }, + "left_excluded": [ + "CHEBI:57540", + "CHEBI:59895" + ], + "reac_id": "RHEA:27433", + "rel_direction": 1, + "right": { + "CHEBI:59896": 1 + }, + "right_excluded": [ + "CHEBI:15378", + "CHEBI:57945" + ], + "rule_id": "RR:03-FFFFD8-869728-14F123", + "rule_score": 0.8408964, + "subs_id": "CHEBI:59895" + }, + "RHEA:27437": { + "left": { + "CHEBI:59895": 1 + }, + "left_excluded": [ + "CHEBI:16240", + "CHEBI:59895" + ], + "reac_id": "RHEA:27437", + "rel_direction": 1, + "right": { + "CHEBI:59896": 1 + }, + "right_excluded": [ + "CHEBI:15377", + "CHEBI:15377" + ], + "rule_id": "RR:03-FFFFD8-869728-14F123", + "rule_score": 0.8408964, + "subs_id": "CHEBI:59895" + } + } +} diff --git a/tests/test_rrCache.py b/tests/test_rrCache.py index 20615b9..8ef5d76 100644 --- a/tests/test_rrCache.py +++ b/tests/test_rrCache.py @@ -1,149 +1,187 @@ """ -Created on Jul 15 2020 +Pytest rewrite of the original unittest-based rrCache tests. -@author: Joan Hérisson +This version uses native pytest parametrization so each cspace/case is +collected and reported as a separate test item instead of being grouped under +unittest subTest blocks. """ -# import logging -from unittest import TestCase -from os import ( - remove as os_rm, - path as os_path -) -from logging import Logger +from __future__ import annotations + from json import load as json_load +from logging import Logger +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 -from brs_utils import ( - create_logger, - extract_gz, - check_file_size -) HERE = os_path.dirname(os_path.abspath(__file__)) -DATA_PATH = os_path.join(HERE, 'data') - - -class Test_rrCache(TestCase): - - cspace = 'mnx3.1' - outdir = f'cache-{cspace}' - cache = rrCache(cspace=cspace, interactive=False) - - # Not possible to compare hashes since - # files contain dict that have to be sorted - # before comparing them and then fill up the memory - # Size of gunzipped files - - def setUp(self, logger: Logger = None): - if logger is None: - self.logger = create_logger(__name__, 'ERROR') - else: - self.logger = logger - # Set attributes from data files - for elem in ['reactions', 'retrorules']: - with open(os_path.join(DATA_PATH, f'{elem}.json'), 'r') as f: - setattr(self, f'{elem}', json_load(f)) - for elem in ['compounds', 'metrics']: - with open(os_path.join(DATA_PATH, f'{elem}_{self.cspace}.json'), 'r') as f: - setattr(self, f'{elem}', json_load(f)) - - def test_all_attr(self): - r"""Test of loading all attributes in rrCache and store them in files. - - Method: Load a full rrCache in 'file' store mode. Then, for each - attribute, compare its length with it is supposed to be. - """ - cache = rrCache(cspace=self.cspace, interactive=False, logger=self.logger) - for attr in self.metrics: - length = self.metrics[attr]['length'] - with self.subTest(attr=attr, length=length): - self.assertEqual(len(cache.get(attr)), length) - - def test_single_attr_file(self): - r"""Test of loading each attribute in rrCache and store it in a file. - - Method: Load a rrCache in 'file' store mode for each single attribute. - Then, compare its length with it is supposed to be. - """ - for attr in self.metrics: - length = self.metrics[attr]['length'] - with self.subTest(attr=attr, length=length): - cache = rrCache(cspace=self.cspace, interactive=False, logger=self.logger) - self.assertEqual(len(cache.get(attr)), length) - - def test_generate_cache(self): - r"""Test of genrating all rrCache files from input_cache. - - Method: Generate a full rrCache. Then, for each file, compare its size - with it is supposed to be. - """ - self.skipTest("Too long, not in standard tests") - rrCache.generate_cache(self.outdir, interactive=False, logger=self.logger) - for name in self.metrics: - filepath = os_path.join(self.outdir, f'{name}.json.gz') - outfile = extract_gz(filepath, self.outdir) - self.assertTrue( - check_file_size( - outfile, - self.metrics[name]['file_size'], - self.logger - ) - ) - os_rm(outfile) - - def test_get_compound(self): - self.assertDictEqual( - self.cache.get_compound('MNXM2'), - self.compounds['MNXM2'] +DATA_PATH = os_path.join(HERE, "data") +# CSPACES = ["mnx3.1", "mnx4.4", "rr2026-v3.0","rr2026-v3.1.0"] +CSPACES = ["rr2026-v3.0"] +DATABASES = ["metanetx", "rhea"] +DATASETS = ["compounds", "metrics", "retrorules", "reactions"] + + +def _load_json(filepath: str): + with open(filepath, "r") as handle: + return json_load(handle) + + +# Load reference data at collection time so pytest can parametrize individual +# cases and report them separately. +REFERENCE_DATA = { + cspace: { + dataset: _load_json(os_path.join(DATA_PATH, f"{dataset}_{cspace}.json")) + for dataset in DATASETS + } + for cspace in CSPACES +} + +ALL_ATTR_CASES = [ + pytest.param(cspace, attr, metric["length"], id=f"{cspace}-{attr}") + for cspace in CSPACES + for attr, metric in REFERENCE_DATA[cspace]["metrics"].items() +] + +COMPOUND_CASES = [ + pytest.param( + cspace, + next(iter(REFERENCE_DATA[cspace]["compounds"])), + id=f"{cspace}-compound", + ) + for cspace in CSPACES +] + +REACTION_CASES = [ + pytest.param( + cspace, + next(iter(REFERENCE_DATA[cspace]["reactions"])), + id=f"{cspace}-reaction", + ) + for cspace in CSPACES +] + +RULE_CASES = [ + pytest.param( + cspace, + next(iter(REFERENCE_DATA[cspace]["retrorules"])), + id=f"{cspace}-rule", + ) + for cspace in CSPACES +] + + +@pytest.fixture(scope="session") +def logger() -> Logger: + return create_logger(__name__, "ERROR") + + +@pytest.fixture(scope="session") +def reference_data() -> dict[str, dict[str, dict]]: + return REFERENCE_DATA + + +@pytest.fixture(scope="session") +def caches(logger: Logger) -> dict[str, rrCache]: + return { + cspace: rrCache( + cspace=cspace, databases=DATABASES, interactive=False, logger=logger ) + for cspace in CSPACES + } - def test_get_list_of_compounds(self): - self.assertTrue( - 'MNXM2' in self.cache.get_list_of_compounds() - ) - self.assertEqual( - len(self.cache.get_list_of_compounds()), - self.metrics['cid_strc']['length'] - ) - def test_get_reaction(self): - self.assertDictEqual( - self.cache.get_reaction('MNXR94688'), - self.reactions['MNXR94688'] - ) - self.assertDictEqual( - self.cache.get_reaction('MNXR94688'), - self.reactions['MNXR94688'] - ) +@pytest.mark.parametrize("cspace, attr, expected_length", ALL_ATTR_CASES) +def test_all_attr(caches, cspace: str, attr: str, expected_length: int): + """Load all rrCache attributes and validate their expected lengths.""" + assert len(caches[cspace].get(attr)) == expected_length - def test_get_list_of_reactions(self): - self.assertTrue( - 'MNXR94688' in self.cache.get_list_of_reactions() - ) - self.assertEqual( - len(self.cache.get_list_of_reactions()), - self.metrics['template_reactions']['length'] - ) - self.assertTrue( - 'MNXR94688' in self.cache.get_list_of_reactions() - ) - self.assertEqual( - len(self.cache.get_list_of_reactions()), - self.metrics['template_reactions']['length'] - ) +@pytest.mark.skip(reason="Too long, not in standard tests") +@pytest.mark.parametrize("cspace", CSPACES, ids=CSPACES) +def test_generate_cache(cspace: str, logger: Logger, reference_data): + """Generate rrCache files and validate the extracted file sizes.""" + outdir = f"cache-{cspace}" + rrCache.generate_cache(outdir, interactive=False, logger=logger) + metrics = reference_data[cspace]["metrics"] - def test_get_reaction_rule(self): - self.assertDictEqual( - self.cache.get_reaction_rule('RR-02-f85f00f767901186-16-F'), - self.retrorules['RR-02-f85f00f767901186-16-F'] - ) + for name, meta in metrics.items(): + filepath = os_path.join(outdir, f"{name}.json.gz") + outfile = extract_gz(filepath, outdir) + try: + assert check_file_size(outfile, meta["file_size"], logger) + finally: + if os_path.exists(outfile): + os_rm(outfile) - def test_get_list_of_reaction_rules(self): - self.assertTrue( - 'RR-02-f85f00f767901186-16-F' in self.cache.get_list_of_reaction_rules() - ) - self.assertEqual( - len(self.cache.get_list_of_reaction_rules()), - self.metrics['rr_reactions']['length'] - ) + +@pytest.mark.parametrize("cspace, cmpd_id", COMPOUND_CASES) +def test_get_compound(caches, reference_data, cspace: str, cmpd_id: str): + compounds = reference_data[cspace]["compounds"] + compound = caches[cspace].get_compound(cmpd_id) + compound["xref"] = caches[cspace].get_compound_xref(cmpd_id) + print(compound) + print() + print(compounds[cmpd_id]) + assert compound == compounds[cmpd_id] + + +@pytest.mark.parametrize("cspace, cmpd_id", COMPOUND_CASES) +def test_get_list_of_compounds(caches, reference_data, cspace: str, cmpd_id: str): + compound_ids = caches[cspace].get_list_of_compounds() + expected_length = reference_data[cspace]["metrics"]["cid_strc"]["length"] + + assert cmpd_id in compound_ids + assert len(compound_ids) == expected_length + + +@pytest.mark.parametrize("cspace, rxn_id", REACTION_CASES) +def test_get_reaction(caches, reference_data, cspace: str, rxn_id: str): + reactions = reference_data[cspace]["reactions"] + assert caches[cspace].get_reaction(rxn_id) == reactions[rxn_id] + + +@pytest.mark.parametrize("cspace, rxn_id", REACTION_CASES) +def test_get_list_of_reactions(caches, reference_data, cspace: str, rxn_id: str): + reaction_ids = caches[cspace].get_list_of_reactions() + expected_length = reference_data[cspace]["metrics"]["template_reactions"]["length"] + + assert rxn_id in reaction_ids + assert len(reaction_ids) == expected_length + + +@pytest.mark.parametrize("cspace, rule_id", RULE_CASES) +def test_get_reaction_rule(caches, reference_data, cspace: str, rule_id: str): + retrorules = reference_data[cspace]["retrorules"] + assert caches[cspace].get_reaction_rule(rule_id) == retrorules[rule_id] + + +@pytest.mark.parametrize("cspace, cmpd_id", COMPOUND_CASES) +def test_contains_compound_in_cache(caches, cspace: str, cmpd_id: str): + assert cmpd_id in caches[cspace] + assert "NOT_A_VALID_ID" not in caches[cspace] + + +@pytest.mark.parametrize("cspace, rxn_id", REACTION_CASES) +def test_contains_reaction_in_cache(caches, cspace: str, rxn_id: str): + assert rxn_id in caches[cspace] + assert "NOT_A_VALID_ID" not in caches[cspace] + + +@pytest.mark.parametrize("cspace, rule_id", RULE_CASES) +def test_contains_rule_in_cache(caches, cspace: str, rule_id: str): + assert rule_id in caches[cspace] + assert "NOT_A_VALID_ID" not in caches[cspace] + + +@pytest.mark.parametrize("cspace, rule_id", RULE_CASES) +def test_get_list_of_reaction_rules(caches, reference_data, cspace: str, rule_id: str): + rule_ids = caches[cspace].get_list_of_reaction_rules() + expected_length = reference_data[cspace]["metrics"]["rr_reactions"]["length"] + + assert rule_id in rule_ids + assert len(rule_ids) == expected_length