Merge pull request #79 from sudlab/ns-rse/72-default-config

.pre-commit-config.yaml

@@ -81,7 +81,7 @@ repos:
       - id: rst-inline-touching-normal
       - id: python-no-eval
       - id: python-check-blanket-noqa
-      - id: python-check-blanket-type-ignore
+      # - id: python-check-blanket-type-ignore
       - id: python-check-mock-methods
 
   - repo: https://github.com/pre-commit/mirrors-mypy

isoslam/default_config.yaml

@@ -0,0 +1,9 @@
+# Default configuration file for running IsoSlam
+log_level: "warning"
+output_dir: output
+bam_file: data/bam/.bam
+gtf_file: data/gtf/
+bed_file: data/bed/
+vcf_file: data/vcf/
+summary: false
+plot: false

isoslam/io.py

@@ -0,0 +1,177 @@
+"""Module for reading files."""
+
+import argparse
+from datetime import datetime
+from importlib import resources
+from pathlib import Path
+
+# from cgatcore import iotools
+from loguru import logger
+
+# import pysam
+from ruamel.yaml import YAML, YAMLError
+
+CONFIG_DOCUMENTATION_REFERENCE = """# For more information on configuration and how to use it see:
+# https://sudlab.github.io/IsoSLAM\n"""
+
+
+def _get_date_time() -> str:
+    """
+    Get a date and time for adding to generated files or logging.
+
+    Returns
+    -------
+    str
+        A string of the current date and time, formatted appropriately.
+    """
+    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+
+def _str_to_path(path: str | Path) -> Path:
+    """
+    Ensure path is a Path object.
+
+    Returns the current directory of passed './'.
+
+    Parameters
+    ----------
+    path : str | Path
+        Path to be converted.
+
+    Returns
+    -------
+    Path
+        Pathlib object of supplied path.
+    """
+    return Path().cwd() if path == "./" else Path(path).expanduser()
+
+
+def _path_to_str(config: dict) -> dict:  # type: ignore[type-arg]
+    """
+    Recursively traverse a dictionary and convert any Path() objects to strings for writing to YAML.
+
+    Parameters
+    ----------
+    config : dict
+        Dictionary to be converted.
+
+    Returns
+    -------
+    Dict:
+        The same dictionary with any Path() objects converted to string.
+    """
+    for key, value in config.items():
+        if isinstance(value, dict):
+            _path_to_str(value)
+        elif isinstance(value, Path):
+            config[key] = str(value)
+    return config
+
+
+def read_yaml(filename: str | Path) -> dict | None:  # type: ignore[type-arg]
+    """
+    Read a YAML file.
+
+    Parameters
+    ----------
+    filename : Union[str, Path]
+        YAML file to read.
+
+    Returns
+    -------
+    Dict
+        Dictionary of the file.
+    """
+    with Path(filename).open(encoding="utf-8") as f:
+        try:
+            yaml_file = YAML(typ="safe")
+            return yaml_file.load(f)  # type: ignore[no-any-return]
+        except YAMLError as exception:
+            logger.error(exception)
+            return {}
+
+
+def write_yaml(
+    config: dict,  # type: ignore[type-arg]
+    output_dir: str | Path,
+    config_file: str = "config.yaml",
+    header_message: str | None = None,
+) -> None:
+    """
+    Write a configuration (stored as a dictionary) to a YAML file.
+
+    Parameters
+    ----------
+    config : dict
+        Configuration dictionary.
+    output_dir : Union[str, Path]
+        Path to save the dictionary to as a YAML file (it will be called 'config.yaml').
+    config_file : str
+        Filename to write to.
+    header_message : str
+        String to write to the header message of the YAML file.
+    """
+    output_config = Path(output_dir) / config_file
+    # Revert PosixPath items to string
+    config = _path_to_str(config)
+
+    if header_message:
+        header = f"# {header_message} : {_get_date_time()}\n" + CONFIG_DOCUMENTATION_REFERENCE
+    else:
+        header = f"# Configuration from IsoSLAM run completed : {_get_date_time()}\n" + CONFIG_DOCUMENTATION_REFERENCE
+    output_config.write_text(header, encoding="utf-8")
+
+    yaml = YAML(typ="safe")
+    with output_config.open("a", encoding="utf-8") as f:
+        try:
+            yaml.dump(config, f)
+        except YAMLError as exception:
+            logger.error(exception)
+
+
+def create_config(args: argparse.Namespace | None = None) -> None:
+    """
+    Write the default configuration file to disk.
+
+    Parameters
+    ----------
+    args : argparse.Namespace | None
+        Optional arguments to parse.
+    """
+    filename = "config" if args.filename is None else args.filename  # type: ignore [union-attr]
+    output_dir = Path("./") if args.output_dir is None else Path(args.output_dir)  # type: ignore [union-attr]
+    output_dir.mkdir(parents=True, exist_ok=True)
+    config_path = resources.files(__package__) / "default_config.yaml"
+    config = config_path.read_text()
+
+    if ".yaml" not in str(filename) and ".yml" not in str(filename):
+        create_config_path = output_dir / f"{filename}.yaml"
+    else:
+        create_config_path = output_dir / filename
+
+    with create_config_path.open("w", encoding="utf-8") as f:
+        f.write(f"# Config file generated {_get_date_time()}\n")
+        f.write(f"{CONFIG_DOCUMENTATION_REFERENCE}")
+        f.write(config)
+    logger.info(f"A sample configuration file has been written to : {str(create_config_path)}")
+    logger.info(CONFIG_DOCUMENTATION_REFERENCE)
+
+
+def load_bam() -> None:
+    """Load '.bam' file."""
+    return
+
+
+def load_bed() -> None:
+    """Load '.bed' file."""
+    return
+
+
+def load_gtf() -> None:
+    """Load '.bed' file."""
+    return
+
+
+def load_vcf() -> None:
+    """Load '.vcf' file."""
+    return

isoslam/processing.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 # from isoslam import __version__
+from isoslam import io
 
 
 def create_parser() -> arg.ArgumentParser:
@@ -110,7 +111,7 @@ def create_parser() -> arg.ArgumentParser:
         default="./",
         help="Path to where the YAML file should be saved (default './' the current directory).",
     )
-    # create_config_parser.set_defaults(func=create_config)
+    create_config_parser.set_defaults(func=io.create_config)
 
     # Additional parsers for future functionality
     # summarize_counts_parser = subparsers.add_parser(
@@ -128,7 +129,7 @@ def create_parser() -> arg.ArgumentParser:
     return parser
 
 
-def process(args: arg.Namespace | None) -> None:
+def process(args: arg.Namespace | None) -> None:  # pylint: disable=unused-argument
     """
     Process a set of files.
 
@@ -142,6 +143,7 @@ def process(args: arg.Namespace | None) -> None:
     None
         Function does not return anything.
     """
+    # config = io.read_yaml() if args.config is None else io.read_yaml(args.config) # type: ignore[call-arg,union-attr]
     return
 
 
@@ -168,7 +170,6 @@ def entry_point(manually_provided_args: list[Any] | None = None, testing: bool =
     args = parser.parse_args() if manually_provided_args is None else parser.parse_args(manually_provided_args)
 
     # If no module has been specified print help and exit
-    print(f"{args.program=}")
     if not args.program:
         parser.print_help()
         sys.exit()

isoslam/utils.py

@@ -0,0 +1,41 @@
+"""Utilities and helper functionsfor IsoSLAM."""
+
+from argparse import Namespace
+
+from loguru import logger
+
+from isoslam import io
+
+
+def update_config(config: dict, args: dict | Namespace) -> dict:  # type: ignore[type-arg]
+    """
+    Update the configuration with any arguments.
+
+    Parameters
+    ----------
+    config : dict
+        Dictionary of configuration (typically read from YAML file specified with '-c/--config <filename>').
+    args : Namespace
+        Command line arguments.
+
+    Returns
+    -------
+    dict
+        Dictionary updated with command arguments.
+    """
+    args = vars(args) if isinstance(args, Namespace) else args
+
+    config_keys = config.keys()
+    for arg_key, arg_value in args.items():
+        if isinstance(arg_value, dict):
+            update_config(config, arg_value)
+        else:
+            if arg_key in config_keys and arg_value is not None:
+                original_value = config[arg_key]
+                config[arg_key] = arg_value
+                logger.debug(f"Updated config config[{arg_key}] : {original_value} > {arg_value} ")
+    if "base_dir" in config.keys():
+        config["base_dir"] = io._str_to_path(config["base_dir"])  # pylint: disable=protected-access
+    if "output_dir" in config.keys():
+        config["output_dir"] = io._str_to_path(config["output_dir"])  # pylint: disable=protected-access
+    return config

pyproject.toml

@@ -36,14 +36,15 @@ keywords = [
 ]
 dependencies = [
   "apsw",
-  "cgat",
-  "cgatcore",
+  # "cgat",
+  # "cgatcore",
   "gevent",
   "loguru",
   "matplotlib",
   "numpy",
   "pandas",
   "pysam",
+  "ruamel.yaml",
   # "ruffus",
   "sqlalchemy",
 ]
@@ -272,7 +273,7 @@ exclude = [
 ]
 
 [[tool.mypy.overrides]]
-module = [ "numpy.*", ]
+module = [ "numpy.*", "loguru", "ruamel.yaml"]
 ignore_missing_imports = true
 
 [project.scripts]

tests/resources/test.yaml

@@ -0,0 +1,12 @@
+this: is
+a: test
+yaml: file
+int: 123
+float: 3.1415
+logical: true
+nested:
+  something: else
+a_list:
+  - 1
+  - 2
+  - 3