diff --git a/README.md b/README.md index d888147..f0028f1 100644 --- a/README.md +++ b/README.md @@ -2,12 +2,12 @@ [![Anaconda-Server Badge](https://anaconda.org/conda-forge/retropath2_wrapper/badges/version.svg)](https://anaconda.org/conda-forge/retropath2_wrapper) [![Anaconda-Server Badge](https://anaconda.org/conda-forge/retropath2_wrapper/badges/latest_release_date.svg)](https://anaconda.org/conda-forge/retropath2_wrapper) -Implementation of the KNIME retropath2.0 workflow. Takes for input the minimal (dmin) and maximal (dmax) diameter for the reaction rules and the maximal path length (maxSteps). The tool expects the following files: `rules.csv`, `sink.csv` and `source.csv` and produces results in an output folder. +Implementation of the KNIME retropath2.0 workflow. Takes for input the minimal (dmin) and maximal (dmax) diameter for the reaction rules and the maximal path length (maxSteps). The tool expects the following files: `rules.csv`, `sink.csv` and `source.csv` and produces results in an output folder. ## Prerequisites -* Python 3 -* KNIME (code was tested on `4.6.4`, `4.7.0` versions) +- Python 3 +- KNIME (code was tested on `4.6.4`, `4.7.0` versions) ## Install @@ -20,6 +20,7 @@ The tool tries to install the KNIME Anlytical Platform as well as the RetroPath2 ### conda package Install in the `` conda environment: + ```shell conda install -c conda-forge -n retropath2_wrapper ``` @@ -29,6 +30,7 @@ conda install -c conda-forge -n retropath2_wrapper **Disclaimer**: we recommand to provide absolute path to files, problems can arise with relative paths. ### From CLI (Linux, macOS) + ```sh python -m retropath2_wrapper --source_file ``` @@ -36,6 +38,7 @@ python -m retropath2_wrapper --source_file export LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$CONDA_PREFIX/lib" ``` ## CI/CD -For further tests and development tools, a CI toolkit is provided in `cicd-toolkit` folder (see [cicd-toolkit/README.md](cicd-toolit/README.md)). +For further tests and development tools, a CI toolkit is provided in `cicd-toolkit` folder (see [cicd-toolkit/README.md](cicd-toolit/README.md)). ### How to cite RetroPath2.0? + Please cite: Delépine B, Duigou T, Carbonell P, Faulon JL. RetroPath2.0: A retrosynthesis workflow for metabolic engineers. Metabolic Engineering, 45: 158-170, 2018. DOI: https://doi.org/10.1016/j.ymben.2017.12.002 diff --git a/environment.macos-arm64.yml b/environment.macos-arm64.yml new file mode 100644 index 0000000..1159759 --- /dev/null +++ b/environment.macos-arm64.yml @@ -0,0 +1,9 @@ +name: retropath2_wrapper +channels: + - conda-forge +dependencies: + - python >=3.10 + - brs_utils >=1.23.1 + - filetype + - colored + - freetype \ No newline at end of file diff --git a/retropath2_wrapper/Args.py b/retropath2_wrapper/Args.py index 288ca6a..20cbb00 100644 --- a/retropath2_wrapper/Args.py +++ b/retropath2_wrapper/Args.py @@ -15,37 +15,10 @@ ) DEFAULTS = { 'MSC_TIMEOUT': 10, # minutes - 'KNIME_VERSION': "4.6.4", 'RP2_VERSION': 'r20250728', 'KNIME_FOLDER': __PACKAGE_FOLDER, - 'KNIME_REPOS': [ - # 'http://update.knime.com/partner/', - 'http://update.knime.com/community-contributions/trunk/', - 'http://update.knime.com/community-contributions/trusted/4.6', - 'http://update.knime.com/analytics-platform/4.6' - ], - 'KNIME_PLUGINS': ','.join( - [ - 'org.knime.base', - 'org.knime.python.nodes', - 'org.knime.datageneration', - 'org.knime.chem.base', - 'org.rdkit.knime.feature.feature.group/4.8.1.v202312052327', - 'org.rdkit.knime.nodes/4.8.1.v202312052327', - # 'org.knime.python.nodes/4.6.0.v202203011403', - # 'org.knime.datageneration/4.6.0.v202202251621', - # 'org.knime.chem.base/4.6.0.v202202251610', - # 'org.rdkit.knime.feature.feature.group/4.8.1.v202312052327', - # 'org.rdkit.knime.nodes/4.8.1.v202312052327', - ] - ), - 'NO_NETWORK': False, "STD_HYDROGEN": "auto", # How hydrogens are represented in chemical rules } -# DEFAULTS['KNIME_PLUGINS'] = ','.join( -# [pkg.split('/')[0] for pkg in DEFAULTS['KNIME_PLUGINS'].split(',')] -# ) -KNIME_ZENODO = {"4.6.4": "7515771", "4.7.0": "7564938"} # Map to Zenodo ID RETCODES = { 'OK': 0, 'NoError': 0, @@ -57,6 +30,7 @@ 'OSError': 2, 'InChI': 3, 'SinkFileMalformed': 4, + 'KnimeInstallationError': 5, } @@ -109,34 +83,11 @@ def _add_arguments(parser): # Knime parser_knime = parser.add_argument_group("Knime arguments") - parser_knime.add_argument( - '--kexec', - type=str, - default=None, - help='path to KNIME executable file (KNIME will be \ - downloaded if not already installed or path is \ - wrong).' - ) parser_knime.add_argument( '--kinstall', type=str, default=DEFAULTS['KNIME_FOLDER'], - help='path to KNIME executable file (KNIME will be \ - downloaded if not already installed or path is \ - wrong).' - ) - parser_knime.add_argument( - '--kver', - type=str, - default=DEFAULTS['KNIME_VERSION'], - choices=list(KNIME_ZENODO.keys()), - help='version of KNIME (mandatory if --kexec is passed).', - ) - parser_knime.add_argument( - '--kplugins', - type=str, - default="", - help='KNIME plugins to use (separated by a comma).', + help='Directory where to find a KNIME executable file', ) # RetroPath2.0 workflow options @@ -149,14 +100,6 @@ def _add_arguments(parser): help=f'version of RetroPath2.0 workflow (default: {DEFAULTS["RP2_VERSION"]}).' ) - # No network option - parser_rp.add_argument( - '--no-network', - action='store_true', - default=DEFAULTS['NO_NETWORK'], - help='Do not use network.' - ) - parser_rp.add_argument('--max_steps' , type=int, default=3) parser_rp.add_argument('--topx' , type=int, default=100) parser_rp.add_argument('--dmin' , type=int, default=0) diff --git a/retropath2_wrapper/RetroPath2.py b/retropath2_wrapper/RetroPath2.py index 16d70ad..45a155c 100644 --- a/retropath2_wrapper/RetroPath2.py +++ b/retropath2_wrapper/RetroPath2.py @@ -45,11 +45,7 @@ def retropath2( rules_file: str, outdir: str, std_hydrogen: str, - kinstall: str = DEFAULTS['KNIME_FOLDER'], - kexec: str = None, - kver: str = DEFAULTS['KNIME_VERSION'], - knime: Knime = None, - kplugins: list = DEFAULTS['KNIME_PLUGINS'], + knime: Knime, rp2_version: str = DEFAULTS['RP2_VERSION'], max_steps: int = 3, topx: int = 100, @@ -65,9 +61,6 @@ def retropath2( logger.debug(f'rules_file: {rules_file}') logger.debug(f'outdir: {outdir}') logger.debug(f'std_hydrogen: {std_hydrogen}') - logger.debug(f'kexec: {kexec}') - logger.debug(f'kinstall: {kinstall}') - logger.debug(f'kver: {kver}') logger.debug(f'rp2_version: {rp2_version}') logger.debug(f'max_steps: {max_steps}') logger.debug(f'topx: {topx}') @@ -78,12 +71,15 @@ def retropath2( # Create Knime object if knime is None: - knime = Knime(kexec=kexec, kinstall=kinstall, kver=kver) + knime = Knime() if rp2_version is not None: knime.workflow = os_path.join( here, 'workflows', f'RetroPath2.0_{rp2_version}.knwf' ) - + if knime.kexec == "": + # Install KNIME + if not knime.install(kver=Knime.DEFAULT_VERSION, logger=logger): + return RETCODES["KnimeInstallationError"] logger.debug('knime: ' + str(knime)) # Store RetroPath2 params into a dictionary @@ -101,11 +97,6 @@ def retropath2( if r_code != RETCODES['OK']: return r_code, None - # Install KNIME - r_code = knime.install(logger=logger) - if r_code > 0: - return r_code, None - logger.info('{attr1}Initializing{attr2}'.format(attr1=attr('bold'), attr2=attr('reset'))) # Preferences diff --git a/retropath2_wrapper/__main__.py b/retropath2_wrapper/__main__.py index 63c5f20..6c3a1b7 100644 --- a/retropath2_wrapper/__main__.py +++ b/retropath2_wrapper/__main__.py @@ -79,18 +79,11 @@ def _cli(): # Create Knime object here = os_path.dirname(os_path.realpath(__file__)) - if args.kplugins == "": - kplugins = [] - else: - kplugins = args.kplugins.split(',') knime = Knime( - kexec=args.kexec, kinstall=args.kinstall, - kver=args.kver, - kplugins=kplugins, workflow=os_path.join(here, 'workflows', 'RetroPath2.0_%s.knwf' % (args.rp2_version,)), - network=not args.no_network, ) + # Print out configuration if not args.silent and args.log.lower() not in ['critical', 'error']: print_conf(knime, prog = parser.prog) @@ -194,12 +187,6 @@ def parse_and_check_args( args = parser.parse_args() - if args.kexec is not None: - if not os.path.isfile(args.kexec): - parser.error("--kexec is not a file: %s" %(args.kexec,)) - if not os.access(args.kexec, os.X_OK): - parser.error("--kexec is not executable: %s" %(args.kexec,)) - # Create outdir if does not exist if not os_path.exists(args.outdir): os_mkdir(args.outdir) diff --git a/retropath2_wrapper/knime.py b/retropath2_wrapper/knime.py index e99fe90..7bfc5fe 100644 --- a/retropath2_wrapper/knime.py +++ b/retropath2_wrapper/knime.py @@ -1,3 +1,5 @@ +import argparse +import glob import os import requests import shutil @@ -5,6 +7,7 @@ import sys import tempfile import urllib.parse +from pathlib import Path from getpass import getuser from logging import ( getLogger, @@ -13,7 +16,7 @@ ) from typing import Any, Dict, Optional from colored import attr -from typing import List +from typing import Set from subprocess import PIPE as sp_PIPE from brs_utils import ( @@ -25,182 +28,68 @@ ) from retropath2_wrapper.Args import ( DEFAULTS, - KNIME_ZENODO, RETCODES, ) from retropath2_wrapper.preference import Preference -class KPlugin(): - """ A Knime package/plugin class""" - - def __init__(self, name: str): - name_version = name.split("/") - self.name = name_version[0] - self.version = name_version[1] if len(name_version) > 1 else "" - - def __repr__(self): - if self.version != "": - return f"{self.name}/{self.version}" - return f"{self.name}" - - def __eq__(self, other): - return self.name == other.name and self.version == other.version - - def __lt__(self, other): - return self.version < other.version - - def __hash__(self): - return hash(self.name) - - def has_version(self): - return self.version != "" - - class Knime(object): """Knime is useful to install executable, install packages or commandline. Attributes ---------- + kinstall: str + directory to found a "knime" executable workflow: str path of the Knime workflow - kver: str - knime version to download, install or use - kexec: str - path of Knime executable - kexec_install: bool - install or not Knime executable - kinstall: str - path install knime - kurl: str - an url to download Knime (from Knime or Zenodo) - kzenodo_id: str - Zenodo repository ID - - Methods - ------- - zenodo_show_repo(self) -> Dict[str, Any] - Show Zenodo repository informations. - - @classmethod - - def standardize_path(cls, path: str) -> str - Path are given with double backslashes on windows. - - install_exec(self, logger: Logger = getLogger(__name__)) -> None - Install Knime executable - - install_pkgs(self, logger: Logger = getLogger(__name__)) -> int - Install KNIME packages needed to execute RetroPath2.0 workflow. - - call(self, files: Dict, params: Dict, preference: Preference, logger: Logger = getLogger(__name__)) -> int - Run Knime workflow. """ ZENODO_API = "https://zenodo.org/api/" KNIME_URL = "http://download.knime.org/analytics-platform/" - + ZENODO = { + "4.6.4": "7515771", + "4.7.0": "7564938", + } + DEFAULT_VERSION = "4.6.4" + PLUGINGS = [ + "org.eclipse.equinox.preferences", + "org.knime.chem.base", + "org.knime.datageneration", + "org.knime.features.chem.types.feature.group", + "org.knime.features.datageneration.feature.group", + "org.knime.features.python.feature.group", + "org.knime.python.nodes", + "org.rdkit.knime.feature.feature.group", + "org.rdkit.knime.nodes", + ] def __init__( self, - workflow: str = "", kinstall: str = DEFAULTS['KNIME_FOLDER'], - kver: str = DEFAULTS['KNIME_VERSION'], - kexec: Optional[str] = None, - kplugins: Optional[str] = DEFAULTS['KNIME_PLUGINS'], - network: bool = not DEFAULTS['NO_NETWORK'], - *args, - **kwargs + workflow: str = "", ) -> None: - - self.workflow = workflow self.kinstall = kinstall - self.kver = kver - self.kexec = kexec - self.kexec_install = False - self.kpkg_install = "" - self.kurl = "" - self.kzenodo_id = "" - self.network = network - - # Setting kexec, kpath, kinstall, kver - if self.kexec is None: - - if not self.network: - raise ValueError('Network is disabled (--no-network) and no KNIME executable is provided (--kexec).') - - self.kzenodo_id = KNIME_ZENODO[self.kver] - zenodo_query = self.__zenodo_show_repo() - for zenodo_file in zenodo_query["files"]: - if sys.platform in zenodo_file["links"]["self"]: - self.kurl = zenodo_file["links"]["self"] - break - - self.kinstall = os.path.join(self.kinstall, '.knime', sys.platform) - if sys.platform == 'darwin': - kpath = os.path.join(self.kinstall, f'KNIME_{self.kver}.app') - self.kexec = os.path.join(kpath, 'Contents', 'MacOS', 'knime') - else: - kpath = os.path.join(self.kinstall, f'knime_{self.kver}') - self.kexec = os.path.join(kpath, 'knime') - if sys.platform == 'win32': - self.kexec += '.exe' - - # Check if exec already exists - if not os.path.exists(self.kexec): - self.kexec_install = True - - # Create url - self.kurl = "" - if self.kver != "": - if sys.platform == "linux": - self.kurl = urllib.parse.urljoin(self.KNIME_URL, "linux/knime_%s.linux.gtk.x86_64.tar.gz" % (self.kver,)) - elif sys.platform == "darwin": - self.kurl = urllib.parse.urljoin(self.KNIME_URL, "macosx/knime_%s.app.macosx.cocoa.x86_64.dmg" % (self.kver,)) - else: - self.kurl = urllib.parse.urljoin(self.KNIME_URL, "win/knime_%s.win32.win32.x86_64.zip" % (self.kver,)) - - # Pkg variable - self.kpkg_install = kpath - if sys.platform == 'darwin': - self.kpkg_install = os.path.join(self.kpkg_install, 'Contents', 'Eclipse') - - else: - if sys.platform in ['linux', 'darwin']: - self.kinstall = os.path.dirname(os.path.dirname(self.kexec)) - self.kver = "" - - self.kplugins = kplugins - self.plugins_default = list( - map(KPlugin, DEFAULTS['KNIME_PLUGINS'].split(',')) - ) - + self.workflow = workflow + self.kexec = Knime.find_executable(path=self.kinstall) def __repr__(self): - s = ["Knime vars:"] - s.append("workflow: " + self.workflow) - s.append("kver: " + self.kver) - s.append("kpkg_install: " + self.kpkg_install) - s.append("kexec: " + self.kexec) - s.append("kexec_install: " + str(self.kexec_install)) - s.append("kinstall: " + self.kinstall) - if self.kurl != "": - s.append("kurl: " + self.kurl) - if self.kzenodo_id != "": - s.append("kzenodo_id: " + self.kzenodo_id) + s = [] + s.append(f"workflow: {self.workflow}") + s.append(f"kinstall: {self.kinstall}") + s.append(f"kexec: {self.kexec}") return "\n".join(s) - - def __zenodo_show_repo(self) -> Dict[str, Any]: + @classmethod + def zenodo_show_repo(cls, kver: str) -> Dict[str, Any]: """Show Zenodo repository informations. Return ------ Dict[str, Any] """ - if not self.network: - raise ValueError('Unable to show the zeonodo repo beacause network is disabled (--no-network).') + + kzenodo_id = Knime.ZENODO[kver] url = urllib.parse.urljoin( - self.ZENODO_API, "records/%s" % (self.kzenodo_id,) + Knime.ZENODO_API, f"records/{kzenodo_id}" ) r = requests.get(url) if r.status_code > 202: @@ -208,7 +97,6 @@ def __zenodo_show_repo(self) -> Dict[str, Any]: return r.json() @classmethod - def standardize_path(cls, path: str) -> str: """Path are given with double backslashes on windows. Knime needs a path with simple slash in commandline. @@ -226,258 +114,98 @@ def standardize_path(cls, path: str) -> str: path = "/".join(path.split(os.sep)) return path + @classmethod + def find_executable(cls, path: str) -> str: + for root, _, files in os.walk(path): + for file in files: + path_file = os.path.join(root, file) + if os.access(path_file, os.X_OK) and os.path.isfile(path_file): + if file.lower() == "knime": + return os.path.abspath(path_file) + return "" - def __install_exec(self, logger: Logger = getLogger(__name__)) -> None: - """Install Knime executable - - Return - ------ - None - """ - logger.info('{attr1}Downloading KNIME {kver}...{attr2}'.format(attr1=attr('bold'), kver=self.kver, attr2=attr('reset'))) - if sys.platform == 'linux': - download_and_extract_tar_gz(self.kurl, self.kinstall) - chown_r(self.kinstall, getuser()) - # chown_r(kinstall, geteuid(), getegid()) - elif sys.platform == 'darwin': - with tempfile.NamedTemporaryFile() as tempf: - download(self.kurl, tempf.name) - app_path = f'{self.kinstall}/KNIME_{self.kver}.app' - if os.path.exists(app_path): - shutil.rmtree(app_path) - with tempfile.TemporaryDirectory() as tempd: - cmd = f'hdiutil mount -noverify {tempf.name} -mountpoint {tempd}/KNIME' - subprocess_call(cmd, logger=logger) - shutil.copytree( - f'{tempd}/KNIME/KNIME {self.kver}.app', - app_path - ) - cmd = f'hdiutil unmount {tempd}/KNIME' - subprocess_call(cmd, logger=logger) - else: # Windows - download_and_unzip(self.kurl, self.kinstall) - logger.info(' |--url: ' + self.kurl) - logger.info(' |--install_dir: ' + self.kinstall) - - - def __manage_pkgs( - self, - plugins_to_install: Optional[str] = DEFAULTS['KNIME_PLUGINS'], - plugins_to_remove: Optional[str] = [], - logger: Logger = getLogger(__name__) - ) -> int: - """Install KNIME packages needed to execute RetroPath2.0 workflow. - - Parameters - ---------- - plugins_to_install: Optional[str] - KNIME plugins to install (separated by a comma). - plugins_to_remove: Optional[str] - KNIME plugins to remove (separated by a comma). - logger : Logger - The logger object. - - Return - ------ - int - """ - StreamHandler.terminator = "" - logger.info( ' |- Checking KNIME packages...') - logger.debug(f' + kpkg_install: {self.kpkg_install}') - logger.debug(f' + kver: {self.kver}') - logger.debug(f' + plugins to install: {plugins_to_install}') - logger.debug(f' + plugins to remove: {plugins_to_remove}') - - if plugins_to_install == plugins_to_remove == []: - StreamHandler.terminator = "\n" - logger.info(' OK') - return 0 - - # tmpdir = tempfile.mkdtemp() - # tmpdir = self.kinstall - - args = [self.kexec] - args += ['-application', 'org.eclipse.equinox.p2.director'] - args += ['-nosplash'] - args += ['-consoleLog'] - # # Download from Zenodo - # zenodo_query = self.zenodo_show_repo() - # repositories = [] - # for zenodo_file in zenodo_query["files"]: - # url = zenodo_file["links"]["self"] - # if "update.analytics-platform" in url or "TrustedCommunityContributions" in url: - # repo_path = os.path.join(tmpdir, os.path.basename(url)) - # if not os.path.exists(repo_path): - # logger.info(f' + Downloading {url} to {repo_path}...') - # download(url, repo_path) - # repositories.append(repo_path) - args += ["-repository"] + [','.join(DEFAULTS['KNIME_REPOS'])] - # args.append(",".join(["jar:file:%s!/" % (x,) for x in repositories])) - args += ['-bundlepool', self.kpkg_install] - args += ['-destination', self.kpkg_install] - - _args = [] - if plugins_to_remove: - _args = ['-uninstallIU'] + [','.join(plugins_to_remove)] - # CPE = subprocess_call(" ".join(args+_args), logger=logger) - - if plugins_to_install: - _args = ["-installIU"] + [','.join(plugins_to_install)] - - CPE = subprocess_call(" ".join(args+_args), logger=logger) - - StreamHandler.terminator = "\n" - # shutil.rmtree(tmpdir, ignore_errors=True) - - logger.info(' OK') - return CPE.returncode - - - def install(self, logger: Logger = getLogger(__name__)) -> int: - logger.debug(f'kexec_install: {self.kexec_install}') - - r_code = 0 - - if not self.network: - logger.warning('Unable to install KNIME nor plugins because network is disabled (--no-network).') - return r_code - - # If order to install, install exec and pkgs - if self.kexec_install: - self.__install_exec(logger=logger) - plugins_to_install = self.plugins_default - plugins_to_remove = [] - else: - plugins_installed = self.list_plugins(logger=logger) - plugins_to_install = self.detect_plugins_to_install(plugins_installed, logger) - # Build list of plugins to remove - # build list of plugin names to install - plugins_to_install_names = [pkg.name for pkg in plugins_to_install] - # build list of plugin names already installed - plugins_installed_names = [pkg.name for pkg in plugins_installed] - # build list of plugin names to remove - plugins_to_remove = list( - set(plugins_installed_names).intersection(set(plugins_to_install_names)) - ) - - # transform lists of KPlugins into list of str - plugins_to_install = [str(pkg) for pkg in plugins_to_install] - - r_code = self.__manage_pkgs( - plugins_to_install, - plugins_to_remove, - logger=logger - ) - - return r_code - - - def detect_plugins_to_install( - self, - plugins_installed: List[KPlugin], - logger: Logger = getLogger(__name__) - ) -> List[str]: - """Detect KNIME plugins to install. - - Parameters - ---------- - plugins_installed: List[KPlugin] - List of installed plugins. - logger : Logger - The logger object. - - Return - ------ - List[str] - """ - - # Be sure that plugins - # - specified by the user - # will be installed/updated - plugins_to_install = list( - map(KPlugin, self.kplugins) - ) - - # Add plugins listed by default iff: - # - not already installed - # - AND not specified by the user - plugins_installed_names = [pkg.name for pkg in plugins_installed if pkg.name] - plugins_to_install_names = [pkg.name for pkg in plugins_to_install if pkg.name] - for pkg in self.plugins_default: - if pkg.name not in plugins_installed_names \ - and pkg.name not in plugins_to_install_names: - # add pkg to 'plugins_to_install' - plugins_to_install.append(pkg) - - # Remove duplicates - plugins_to_install = list(set(plugins_to_install)) - - # For plugins appearing several times in 'plugins_to_install', - # keep only the one with the most specified version (e.g. w/ '/4.6.0.v202202251621') - _plugins_to_install = plugins_to_install.copy() - plugins_to_install = [] - for _pkg in _plugins_to_install: - _pkg_name = _pkg.name - # if pkg basename not in 'plugins_to_install' - if _pkg_name not in [ - pkg.name for pkg in plugins_to_install - ]: - # keep pkg in the list of plugins to install - plugins_to_install.append(_pkg) - else: - # check if pkg version is specified - if '/' in _pkg: - # if pkg version is specified, keep it - plugins_to_install.append(_pkg) - - # If plugin version is specified, - # remove it if: - # - it is already installed, - # - AND installed version > specified version - for pkg_to_install in plugins_to_install: - # version is specified - if pkg_to_install.has_version(): - # if pkg is already installed - for pkg_installed in plugins_installed: - if pkg_to_install.name == pkg_installed.name: - if pkg_to_install.version <= pkg_installed.version: - # remove pkg from 'plugins_to_install' - plugins_to_install.remove(pkg_to_install) - break - - return plugins_to_install - - - def list_plugins(self, logger: Logger = getLogger(__name__)) -> list: - """List KNIME plugins. - - Parameters - ---------- - logger : Logger - The logger object. - - Return - ------ - list - """ - args = [self.kexec] - args += ['-application', 'org.eclipse.equinox.p2.director'] - args += ['-nosplash'] - args += ['-consoleLog'] - args += ['-lir'] - args += ['-d', self.kpkg_install] - CPE = subprocess_call( - " ".join(args), - stdout=sp_PIPE, - logger=logger - ) + @classmethod + def find_p2_dir(cls, path: str) -> str: + for dirpath, dirnames, _ in os.walk(path): + if "p2" in dirnames: + return os.path.abspath(os.path.join(dirpath, "p2")) + return "" - return [ - KPlugin(elt) - for elt in CPE.stdout.decode('utf-8').split("\n") - if (elt.startswith('org.') or elt.startswith('com.')) - ] + @classmethod + def collect_top_level_dirs(cls, path) -> Set: + root = Path(path) + names = set() + for p in root.iterdir(): + if p.is_dir(): + names.add(p.name) + return names + + def install(self, kver: str, logger: Logger = getLogger(__name__)) -> bool: + data = Knime.zenodo_show_repo(kver=kver) + # Install Knime + dirs_before = Knime.collect_top_level_dirs(path=self.kinstall) + for file in data["files"]: + basename = file["key"] + url = file["links"]["self"] + if "linux" in basename and sys.platform == "linux": + download_and_extract_tar_gz(url, self.kinstall) + chown_r(self.kinstall, getuser()) + # chown_r(kinstall, geteuid(), getegid()) + break + elif "macosx" in basename and sys.platform == "darwin": + with tempfile.NamedTemporaryFile() as tempf: + download(url, tempf.name) + app_path = f'{self.kinstall}/KNIME_{kver}.app' + if os.path.exists(app_path): + shutil.rmtree(app_path) + with tempfile.TemporaryDirectory() as tempd: + cmd = f'hdiutil mount -noverify {tempf.name} -mountpoint {tempd}/KNIME' + subprocess_call(cmd, logger=logger) + shutil.copytree( + f'{tempd}/KNIME/KNIME {kver}.app', + app_path + ) + cmd = f'hdiutil unmount {tempd}/KNIME' + subprocess_call(cmd, logger=logger) + break + elif "win32" in basename and sys.platform == "win32": + download_and_unzip(url, self.kinstall) + break + dirs_after = Knime.collect_top_level_dirs(path=self.kinstall) + dirs_only_after = dirs_after - dirs_before + assert len(dirs_only_after) == 1, dirs_only_after + + # Download Plugins + tempdir = tempfile.mkdtemp() + try: + path_plugins = [] + for file in data["files"]: + basename = file["key"] + url = file["links"]["self"] + if "org.knime.update" in basename or "TrustedCommunity" in basename: + path_plugin = os.path.join(tempdir, basename) + download(url=url, file=path_plugin) + path_plugins.append(path_plugin) + + # Install Plugins + self.kexec = Knime.find_executable(path=self.kinstall) + p2_dir = Knime.find_p2_dir(path=self.kinstall) + args = [f"{self.kexec}"] + args += ["-nosplash", "-consoleLog"] + args += ["-application", "org.eclipse.equinox.p2.director"] + args += ["-repository", ",".join([f"jar:file:{path_plugin}!/" for path_plugin in path_plugins])] + args += ["-bundlepool", p2_dir] + args += ["-destination", os.path.abspath(os.path.join(self.kinstall, dirs_only_after.pop()))] + args += ["-i", ",".join(Knime.PLUGINGS)] + CPE = subprocess.run(args) + logger.debug(CPE) + except Exception as error: + logger.error(error) + finally: + # Clean up + shutil.rmtree(tempdir) + return True def call( self, @@ -568,3 +296,26 @@ def call( return RETCODES['OSError'] +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument( + "--kinstall", required=True, help="Path to install Knime" + ) + parser.add_argument( + "--kver", default="4.6.4", choices=["4.6.4", "4.7.0"], help="Knime version" + ) + parser.add_argument( + "--overwrite", action="store_true", help="Install even if the executable is not present" + ) + args = parser.parse_args() + + path_knime = args.kinstall + kver = args.kver + to_overwrite = True if args.overwrite else False + + os.makedirs(path_knime, exist_ok=True) + knime = Knime(kinstall=path_knime) + + if to_overwrite or not knime.kexec: + knime.install(kver=kver) + \ No newline at end of file diff --git a/tests/test_knime.py b/tests/test_knime.py index be9ce76..b2059b4 100644 --- a/tests/test_knime.py +++ b/tests/test_knime.py @@ -5,10 +5,7 @@ import tempfile import pytest -from retropath2_wrapper.Args import ( - KNIME_ZENODO, - RETCODES -) +from retropath2_wrapper.Args import RETCODES from retropath2_wrapper.knime import Knime @@ -42,30 +39,11 @@ def test_standardize_path(self): spath = Knime.standardize_path(path=path) assert "\\" not in spath - @pytest.mark.skipif(FUNCTIONAL, reason="Functional test") - def test_install_knime_from_knime(self): - tempdir = tempfile.mkdtemp() - - knime = Knime(workflow="", kinstall=tempdir) - knime.install_exec() - kexec = TestKnime.filter_exec(path=tempdir) - assert kexec is not None - # Failed could be araise due to missing dependecy - try: - ret = knime.install_pkgs() - assert ret == RETCODES['OK'] - except Exception: - pass - shutil.rmtree(tempdir, ignore_errors=True) - @pytest.mark.skipif(FUNCTIONAL, reason="Functional test") def test_install_knime_from_zenodo(self): tempdir = tempfile.mkdtemp() - - knime = Knime(workflow="", kinstall=tempdir, kver=list(KNIME_ZENODO.keys())[0]) - knime.install_exec() + knime = Knime(workflow="", kinstall=tempdir) + knime.install(kver=list(Knime.ZENODO.keys())[0]) kexec = TestKnime.filter_exec(path=tempdir) assert kexec is not None - ret = knime.install_pkgs() - assert ret == RETCODES['OK'] shutil.rmtree(tempdir, ignore_errors=True)