added SkippablePlugin mixin to allow plugins to be disabled/removed from a pipeline (Filter/Writer implement this mixin)

Author: fracpete

CHANGES.rst

@@ -1,6 +1,13 @@
 Changelog
 =========
 
+0.2.16 (2025-04-08)
+-------------------
+
+- filters and writers can be skipped now via the `--skip` flag, making it easy for external
+  scripts to enable/disable pipeline components
+
+
 0.2.15 (2025-03-28)
 -------------------
 

README.md

@@ -57,6 +57,12 @@ the following two methods:
 * `_apply_args(self, ns: argparse.Namespace)`: in this method, you need to 
   transfer the parsed options into member variables of your plugin 
 
+The `SkippablePlugin` mixin, as implemented by `Filter` and `Writer`, allows
+the disabling of the plugin via the `--skip` flag. Plugins that were disabled
+this way, get automatically removed when calling the `args_to_objects` function.
+This is an easy way of turning on/off parts of a pipeline without having to
+remove complex filter/writer configurations.
+
 
 ### Aliases
 

setup.py

@@ -33,7 +33,7 @@ def _read(f):
     install_requires=[
         "setuptools",
     ],
-    version="0.2.15",
+    version="0.2.16",
     author='Peter Reutemann',
     author_email='fracpete@waikato.ac.nz',
     entry_points={

src/seppl/__init__.py

@@ -1,6 +1,7 @@
 from ._plugin import Plugin, PluginWithLogging, OutputProducer, InputConsumer, check_compatibility
 from ._plugin import LoggingHandler, Initializable, init_initializable
 from ._plugin import AliasSupporter, get_aliases, has_aliases, get_all_names
+from ._plugin import SkippablePlugin, add_skip_option
 from ._args import split_cmdline, split_args, args_to_objects, is_help_requested, enumerate_plugins, escape_args, unescape_args, resolve_handler
 from ._registry import Registry, MODE_DYNAMIC, MODE_EXPLICIT, MODES
 from ._class_registry import ClassListerRegistry, get_class_lister, DEFAULT

src/seppl/_args.py

@@ -2,7 +2,7 @@
 import shlex
 
 from typing import List, Dict, Tuple, Iterable, Set, Optional
-from ._plugin import Plugin
+from ._plugin import Plugin, SkippablePlugin
 
 
 def escape_args(args: List[str]) -> List[str]:
@@ -51,6 +51,7 @@ def split_cmdline(cmdline: str, unescape: bool = False) -> List[str]:
         result = unescape_args(result)
     return result
 
+
 def resolve_handler(search: str, handlers: Set[str], partial: bool = False) -> Optional[str]:
     """
     Tries to find the "search" string among the handlers, exact and partial match.
@@ -127,6 +128,7 @@ def split_args(args: List[str], handlers: List[str], unescape: bool = False, par
 def args_to_objects(args: Dict[str, List[str]], valid_plugins: Dict[str, Plugin], allow_global_options: bool = False, allow_unknown_args: bool = False, unescape: bool = False) -> List[Plugin]:
     """
     Instantiates the plugins from the parsed arguments dictionary.
+    Automatically removes SkippablePlugin instances that are to be skipped.
 
     :param args: the arguments dictionary generated by split_args
     :type args: dict
@@ -158,6 +160,18 @@ def args_to_objects(args: Dict[str, List[str]], valid_plugins: Dict[str, Plugin]
         if not allow_unknown_args and (len(unknown) > 0):
             raise Exception("Found unknown argument(s) for plugin '%s': %s" % (plugin.name(), str(unknown)))
         result.append(plugin)
+
+    # remove plugins to be skipped
+    i = 0
+    while i < len(result):
+        try:
+            if isinstance(result[i], SkippablePlugin) and result[i].is_skipped:
+                result.pop(i)
+                continue
+        except:
+            pass
+        i += 1
+
     return result
 
 

src/seppl/_plugin.py

@@ -340,3 +340,31 @@ def check_compatibility(plugins: List[Plugin], match_all=AnyData):
                 "Output(s) of " + index1 + "/" + plugin1.name()
                 + " not compatible with input(s) of " + index2 + "/" + plugin2.name() + ": "
                 + classes_to_str(classes1) + " != " + classes_to_str(classes2))
+
+
+class SkippablePlugin:
+    """
+    Mixin for plugins that can be disabled, i.e., get skipped in the pipeline.
+    """
+
+    @property
+    def is_skipped(self) -> bool:
+        """
+        Returns whether the plugin is to be skipped.
+
+        :return: True if to be skipped
+        :rtype: bool
+        """
+        return False
+
+
+def add_skip_option(parser: argparse.ArgumentParser, help_str: str = "Disables the plugin, removing it from the pipeline."):
+    """
+    Adds the --skip option to the argument parser.
+
+    :param parser: the parser to update
+    :type parser: argparse.ArgumentParser
+    :param help_str: the help string to use
+    :type help_str: str
+    """
+    parser.add_argument("--skip", action="store_true", default=False, help=help_str, required=False)

src/seppl/io/_filter.py

@@ -1,15 +1,16 @@
 import abc
+import argparse
 from typing import List
 
 from wai.logging import LOGGING_WARNING
-from seppl import InputConsumer, OutputProducer, PluginWithLogging, Initializable, init_initializable, Session, SessionHandler
+from seppl import InputConsumer, OutputProducer, PluginWithLogging, Initializable, init_initializable, Session, SessionHandler, SkippablePlugin, add_skip_option
 
 FILTER_ACTION_KEEP = "keep"
 FILTER_ACTION_DISCARD = "discard"
 FILTER_ACTIONS = [FILTER_ACTION_KEEP, FILTER_ACTION_DISCARD]
 
 
-class Filter(PluginWithLogging, InputConsumer, OutputProducer, SessionHandler, Initializable, abc.ABC):
+class Filter(PluginWithLogging, InputConsumer, OutputProducer, SessionHandler, Initializable, SkippablePlugin, abc.ABC):
     """
     Base class for filters.
     """
@@ -26,6 +27,38 @@ def __init__(self, logger_name: str = None, logging_level: str = LOGGING_WARNING
         super().__init__(logger_name=logger_name, logging_level=logging_level)
         self._session = None
         self._last_input = None
+        self.skip = False
+
+    def _create_argparser(self) -> argparse.ArgumentParser:
+        """
+        Creates an argument parser. Derived classes need to fill in the options.
+
+        :return: the parser
+        :rtype: argparse.ArgumentParser
+        """
+        parser = super()._create_argparser()
+        add_skip_option(parser)
+        return parser
+
+    def _apply_args(self, ns: argparse.Namespace):
+        """
+        Initializes the object with the arguments of the parsed namespace.
+
+        :param ns: the parsed arguments
+        :type ns: argparse.Namespace
+        """
+        super()._apply_args(ns)
+        self.skip = ns.skip
+
+    @property
+    def is_skipped(self) -> bool:
+        """
+        Returns whether the plugin is to be skipped.
+
+        :return: True if to be skipped
+        :rtype: bool
+        """
+        return self.skip
 
     @property
     def session(self) -> Session:

src/seppl/io/_writer.py

@@ -1,13 +1,13 @@
 import abc
+import argparse
 from typing import Iterable
 
 from wai.logging import LOGGING_WARNING
 
-from seppl import InputConsumer, PluginWithLogging
-from seppl import SessionHandler, Session
+from seppl import InputConsumer, PluginWithLogging, SessionHandler, Session, SkippablePlugin, add_skip_option
 
 
-class Writer(PluginWithLogging, InputConsumer, SessionHandler, abc.ABC):
+class Writer(PluginWithLogging, InputConsumer, SessionHandler, SkippablePlugin, abc.ABC):
     """
     Ancestor of classes that write data.
     """
@@ -24,6 +24,38 @@ def __init__(self, logger_name: str = None, logging_level: str = LOGGING_WARNING
         super().__init__(logger_name=logger_name, logging_level=logging_level)
         self._session = None
         self._last_input = None
+        self.skip = False
+
+    def _create_argparser(self) -> argparse.ArgumentParser:
+        """
+        Creates an argument parser. Derived classes need to fill in the options.
+
+        :return: the parser
+        :rtype: argparse.ArgumentParser
+        """
+        parser = super()._create_argparser()
+        add_skip_option(parser)
+        return parser
+
+    def _apply_args(self, ns: argparse.Namespace):
+        """
+        Initializes the object with the arguments of the parsed namespace.
+
+        :param ns: the parsed arguments
+        :type ns: argparse.Namespace
+        """
+        super()._apply_args(ns)
+        self.skip = ns.skip
+
+    @property
+    def is_skipped(self) -> bool:
+        """
+        Returns whether the plugin is to be skipped.
+
+        :return: True if to be skipped
+        :rtype: bool
+        """
+        return self.skip
 
     @property
     def session(self) -> Session: