diff --git a/retropath2_wrapper/Args.py b/retropath2_wrapper/Args.py index cb1e239..a5fab8f 100644 --- a/retropath2_wrapper/Args.py +++ b/retropath2_wrapper/Args.py @@ -38,7 +38,8 @@ # 'org.rdkit.knime.feature.feature.group/4.8.1.v202312052327', # 'org.rdkit.knime.nodes/4.8.1.v202312052327', ] - ) + ), + 'NO_NETWORK': False, } # DEFAULTS['KNIME_PLUGINS'] = ','.join( # [pkg.split('/')[0] for pkg in DEFAULTS['KNIME_PLUGINS'].split(',')] @@ -147,6 +148,14 @@ 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/__main__.py b/retropath2_wrapper/__main__.py index dbb56b6..43d014e 100644 --- a/retropath2_wrapper/__main__.py +++ b/retropath2_wrapper/__main__.py @@ -88,6 +88,7 @@ def _cli(): 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']: diff --git a/retropath2_wrapper/knime.py b/retropath2_wrapper/knime.py index 73595b5..9c8681f 100644 --- a/retropath2_wrapper/knime.py +++ b/retropath2_wrapper/knime.py @@ -106,6 +106,7 @@ def __init__( kver: str = DEFAULTS['KNIME_VERSION'], kexec: Optional[str] = None, kplugins: Optional[str] = DEFAULTS['KNIME_PLUGINS'], + network: bool = not DEFAULTS['NO_NETWORK'], *args, **kwargs ) -> None: @@ -118,11 +119,16 @@ def __init__( 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() + 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"] @@ -183,13 +189,15 @@ def __repr__(self): return "\n".join(s) - def zenodo_show_repo(self) -> Dict[str, Any]: + def __zenodo_show_repo(self) -> 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).') url = urllib.parse.urljoin( self.ZENODO_API, "records/%s" % (self.kzenodo_id,) ) @@ -218,7 +226,7 @@ def standardize_path(cls, path: str) -> str: return path - def install_exec(self, logger: Logger = getLogger(__name__)) -> None: + def __install_exec(self, logger: Logger = getLogger(__name__)) -> None: """Install Knime executable Return @@ -251,7 +259,7 @@ def install_exec(self, logger: Logger = getLogger(__name__)) -> None: logger.info(' |--install_dir: ' + self.kinstall) - def manage_pkgs( + def __manage_pkgs( self, plugins_to_install: Optional[str] = DEFAULTS['KNIME_PLUGINS'], plugins_to_remove: Optional[str] = [], @@ -329,6 +337,10 @@ def install(self, logger: Logger = getLogger(__name__)) -> int: 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) @@ -350,7 +362,7 @@ def install(self, logger: Logger = getLogger(__name__)) -> int: # transform lists of KPlugins into list of str plugins_to_install = [str(pkg) for pkg in plugins_to_install] - r_code = self.manage_pkgs( + r_code = self.__manage_pkgs( plugins_to_install, plugins_to_remove, logger=logger