feat: add '--no-network' option

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)

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']:

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