Merge pull request #17 from cokelaer/main

More generic version parsing

README.rst

@@ -40,7 +40,7 @@ Then, just type e.g::
 
     versionix  fastqc
 
-This tool uses a registry so it will work only with resgistered tools, which list can be obtained with::
+This tool uses returns the version as X.Y.Z string. most tools would work. However, a registry is available for complex cases. Registered tools can be obtained with::
 
     versionix --registered
 
@@ -67,6 +67,8 @@ Changelog
 ========= ========================================================================
 Version   Description
 ========= ========================================================================
+0.3.0     Refactor to use regular expression and registry only if needed. This
+          make versionix quite generic.
 0.2.4     More tools in the registry and added precommit
 0.2.3     More tools in the registry
 0.2.2     add all tools required by sequana pipelines (oct 2023)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "versionix"
-version = "0.2.4"
+version = "0.3.0"
 description = "Get version of any tools"
 authors = ["Sequana Team"]
 license = "BSD-3"

test/test_misc.py

@@ -0,0 +1 @@
+from versionix import version

test/test_parser.py

@@ -25,41 +25,46 @@ def test_macs3(fp, mocker):
     assert get_version("macs3") == "3.0.0b1"
 
 
-def test_kallisto(fp, mocker):
-    mocker.patch("shutil.which", return_value="something")
-    fp.register(["kallisto", "version"], stdout=["kallisto, version 0.48.0"])
-    assert get_version("kallisto") == "0.48.0"
-
-
 def test_pigz(fp, mocker):
-    # stderr parser
     mocker.patch("shutil.which", return_value="something")
-    fp.register(["pigz", "--version"], stderr=["pigz 2.4"])
+    fp.register(["pigz", "--version"], stdout=["pigz 2.4"])
     assert get_version("pigz") == "2.4"
 
 
 def test_fastqc(fp, mocker):
-    # stderr parser
     mocker.patch("shutil.which", return_value="something")
-    fp.register(["fastqc", "--version"], stderr=["fastqc v1.0.0"])
+    fp.register(["fastqc", "--version"], stdout=["FastQC v1.0.0"])
     assert get_version("fastqc") == "1.0.0"
 
 
-def _test_bwa(fp, mocker):
+def test_bwa(fp, mocker):
     mocker.patch("shutil.which", return_value="something")
+    fp.register(["bwa", "--version"], stdout=[], stderr=["[main] unrecognized command '--version'"])
+    fp.register(["bwa", "-v"], stdout=[], stderr=["[main] unrecognized command '-v'"])
+    fp.register(["bwa", "version"], stdout=[], stderr=["[main] unrecognized command 'version'"])
+    fp.register(["bwa", "-V"], stdout=[], stderr=["[main] unrecognized command '-V'"])
+    fp.register(["bwa", "-version"], stdout=[], stderr=["[main] unrecognized command '-version'"])
     fp.register(
         ["bwa"],
+        stdout=[],
         stderr=[
             """
+
 Program: bwa (alignment via Burrows-Wheeler transformation)
 Version: 0.7.17-r1188
 Contact: Heng Li <lh3@sanger.ac.uk>
 
 Usage:   bwa <command> [options]
 """
         ],
+        returncode=1,
     )
-    assert get_version("bwa") == "0.7.17-r1188"
+    assert get_version("bwa") == "0.7.17"
+
+
+def test_blacklists(fp, mocker):
+    mocker.patch("shutil.which", return_value="something")
+    assert get_version("Build_Trinotate_Boilerplate_SQLite_db.pl") == "?.?.?"
 
 
 def test_bedtools(fp, mocker):
@@ -76,8 +81,16 @@ def test_bamtools(fp, mocker):
 
 def test_singularity(fp, mocker):
     mocker.patch("shutil.which", return_value="something")
-    fp.register(["singularity", "version"], stdout=["3.6.2+12-gad3457a9a"])
-    assert get_version("singularity") == "3.6.2+12-gad3457a9a"
+    fp.register(["singularity", "--version"], stdout=["apptainer version 1.3.1-1.fc40"])
+    fp.register(["singularity", "version"], stdout=["1.3.1-1.fc403"])
+    assert get_version("singularity") == "1.3.1"
+
+
+def test_registered(fp, mocker):
+    mocker.patch("shutil.which", return_value="something")
+    # fp.register(["dot", "--version"], stdout=["dot - graphviz version 2.40.1 (20161225.0304)"])
+    fp.register(["dot", "-V"], stderr=["dot - graphviz version 2.40.1 (20161225.0304)"])
+    assert get_version("dot") == "2.40.1"
 
 
 def test_script(fp, mocker):
@@ -106,11 +119,3 @@ def test_bedtools_error(fp, mocker):
         assert False
     except SystemExit:
         assert True
-
-
-def test_no_standalone():
-    try:
-        get_version("unknown_tool")
-        assert False
-    except SystemExit:
-        assert True

versionix/__init__.py

@@ -1,6 +1,17 @@
-import pkg_resources
+from importlib import metadata
 
-try:
-    version = pkg_resources.require("versionix")[0].version
-except:  # pragma: no cover
-    version = ">=0.1.0"
+
+def get_package_version(package_name):
+    try:
+        version = metadata.version(package_name)
+        return version
+    except metadata.PackageNotFoundError:  # pragma no cover
+        return f"{package_name} not found"
+
+
+version = get_package_version("versionix")
+
+
+from .logging import Logging
+
+logger = Logging("versionix", "INFO", text_color="green")

versionix/blacklist.py

@@ -0,0 +1 @@
+blacklist = ["Build_Trinotate_Boilerplate_SQLite_db.pl"]

versionix/logging.py

@@ -0,0 +1,121 @@
+# -*- python -*-
+#
+#  This file is part of easydev software
+#
+#  Copyright (c) 2011-2024
+#
+#  File author(s): Thomas Cokelaer <cokelaer@gmail.com>
+#
+#  Distributed under the BSD3 License.
+#
+#  Website: https://github.com/cokelaer/easydev
+#
+##############################################################################
+# import logging
+import colorlog
+
+__all__ = ["Logging"]
+
+
+colors = {
+    "DEBUG": "cyan",
+    "INFO": "green",
+    "WARNING": "yellow",
+    "ERROR": "red",
+    "CRITICAL": "bold_red",
+}
+
+
+class Logging(object):  # pragma no cover
+    """logging utility.
+
+    ::
+
+        >>> l = Logging("root", "INFO")
+        >>> l.info("test")
+        INFO:root:test
+        >>> l.level = "WARNING"
+        >>> l.info("test")
+
+    """
+
+    def __init__(self, name="root", level="WARNING", text_color="blue"):
+        self._name = name
+        self.formatter = colorlog.ColoredFormatter(
+            "%(log_color)s%(levelname)-8s[%(name)s:%(lineno)d]: %(reset)s %({})s%(message)s".format(text_color),
+            datefmt=None,
+            reset=True,
+            log_colors=colors,
+            secondary_log_colors={},
+            style="%",
+        )
+        self._set_name(name)
+
+        logger = colorlog.getLogger(self._name)
+        handler = colorlog.StreamHandler()
+        handler.setFormatter(self.formatter)
+        logger.addHandler(handler)
+
+    def _set_name(self, name):
+        level = self.level
+        self._name = name
+        logger = colorlog.getLogger(self._name)
+        if level == 0:
+            self._set_level("WARNING")
+        else:
+            self._set_level(level)
+
+    def _get_name(self):
+        return self._name
+
+    name = property(_get_name, _set_name)
+
+    def _set_level(self, level):
+        if isinstance(level, bool):
+            if level is True:
+                level = "INFO"
+            if level is False:
+                level = "ERROR"
+        if level == 10:
+            level = "DEBUG"
+        if level == 20:
+            level = "INFO"
+        if level == 30:
+            level = "WARNING"
+        if level == 40:
+            level = "ERROR"
+        if level == 50:
+            level = "CRITICAL"
+        colorlog.getLogger(self.name).setLevel(level)
+
+    def _get_level(self):
+        level = colorlog.getLogger(self.name).level
+        if level == 10:
+            return "DEBUG"
+        elif level == 20:
+            return "INFO"
+        elif level == 30:
+            return "WARNING"
+        elif level == 40:
+            return "ERROR"
+        elif level == 50:
+            return "CRITICAL"
+        else:
+            return level
+
+    level = property(_get_level, _set_level)
+
+    def debug(self, msg):
+        colorlog.getLogger(self.name).debug(msg)
+
+    def info(self, msg):
+        colorlog.getLogger(self.name).info(msg)
+
+    def warning(self, msg):
+        colorlog.getLogger(self.name).warning(msg)
+
+    def critical(self, msg):
+        colorlog.getLogger(self.name).critical(msg)
+
+    def error(self, msg):
+        colorlog.getLogger(self.name).error(msg)

versionix/parser.py

@@ -9,33 +9,107 @@
 #  Website:       https://github.com/sequana/versionix
 #  Contributors:  https://github.com/sequana/versionix/graphs/contributors
 ##############################################################################
-
+import re
 import shutil
 import subprocess
 import sys
 
+import colorlog
+
+from versionix import logger
+
+logger = colorlog.getLogger(logger.name)
+
+
+from .blacklist import blacklist
 from .registry import metadata
 
 
+class Versionix:
+    def __init__(self, standalone):
+
+        self.standalone = standalone
+
+    def get_version(self):
+        """possibilities"""
+        # known options to retrieve version
+        # could be empty (eg ktImportText) from kronatools
+
+        if self.standalone in blacklist:
+            logger.debug(f"{self.standalone} blacklisted")
+            return "?.?.?"
+
+        options = ["--version", "-v", "version", "-V", "-version", ""]
+        for option in options:
+            cmd = f"{self.standalone} {option}".strip()
+            logger.debug(cmd)
+            p = subprocess.run(cmd, capture_output=True, universal_newlines=True, shell=True)
+            if p.returncode == 0:
+                logger.debug(f"return code 0 for {option}")
+                stdout = p.stdout.strip()
+                if stdout:
+                    try:
+                        return self.parse_version(stdout)
+                    except Exception as err:  # pragma: no cover
+                        pass
+            else:
+                logger.debug(f"return code {p.returncode} for {option}")
+                stderr = p.stderr.strip()
+                if stderr:
+                    try:
+                        return self.parse_version(stderr)
+                    except Exception as err:  # pragma no cover
+                        pass
+
+        logger.warning(f"None of {options} looks valid for {self.standalone}")
+        # instead of returning None, we return a string ?.?.? to mimic X.Y.Z pattern
+        return "?.?.?"
+
+    def parse_version(self, string):
+
+        # valir for
+        #    "tools 2.5.2",
+        #    "tools v1.1.0",
+        #    "tools 3.0.0b1",
+        #    "tools, version 0.48.0",
+        #    "tools 2.8"
+        #    "version: v0.9.21
+        # but returns X.Y.Z (or X.Y)
+
+        version = re.search(r"(v?[\d]+\.[\d]+(?:\.[\d]+)?[a-zA-Z0-9]*)", string).group(1)
+        version = version.strip("v")
+        return version
+
+
 def get_version(standalone, verbose=True):
     """Main entry point that returns the version of an existing executable"""
-    # we use check_output in case the standalone opens a GUI (e.g. fastqc --WRONG pops up the GUI)
+    # we should use check_output in case the standalone opens a GUI (e.g. fastqc --WRONG pops up the GUI)
 
-    try:
-        meta = metadata[standalone]
-    except KeyError:
-        if verbose:
-            print(
-                "#ERROR Your input standalone is not registered. Please provide a PR or fill an issue on github/sequana/versionix"
-            )
+    # let us check that the standalone exists locally
+    if shutil.which(standalone) is None:
+        logger.error(f"ERROR: {standalone} command not found in your environment")
         sys.exit(1)
 
-    # If there is no special caller defined, let us check that the standalone exists
-    if "caller" not in meta.keys() and shutil.which(standalone) is None:
-        if verbose:
-            print(f"ERROR: {standalone} command not found in your environment")
+    # is it registered ?
+    if standalone in metadata.keys():
+        version = search_registered(standalone)
+        return version
+
+    # Try a generic search
+    try:
+        v = Versionix(standalone)
+        return v.get_version()
+    except Exception as err:  # pragma: no cover
+        print(err)
         sys.exit(1)
 
+
+def search_registered(standalone):
+
+    # Otherwise, a search using registered names
+    logger.warning(f"Using registered info for {standalone}")
+    meta = metadata[standalone]
+
     # The command used to get the version output
     caller = meta.get("caller", standalone)