Add parse package migrated from bonsai-prp

Author: ryanjameskennedy

src/bonsai_libs/parse/__init__.py

@@ -2,14 +2,7 @@
 
 from importlib import import_module
 from pathlib import Path
-
-from .core.registry import (
-    get_parser,
-    hydrate_result,
-    registered_softwares,
-    registered_version_ranges,
-    run_parser,
-)
+from .core.registry import get_parser, registered_softwares, registered_version_ranges, run_parser, hydrate_result
 
 # auto-import all modules under parse/parsers to ensure that all parsers are registered
 PARSER_DIR = "parsers"
@@ -18,10 +11,4 @@
     if file.name not in ("__init__.py", "utils.py"):
         import_module(f"{__name__}.{PARSER_DIR}.{file.stem}")
 
-__all__ = [
-    "get_parser",
-    "registered_softwares",
-    "registered_version_ranges",
-    "run_parser",
-    "hydrate_result",
-]
+__all__ = ["get_parser", "registered_softwares", "registered_version_ranges", "run_parser", "hydrate_result"]

src/bonsai_libs/parse/core/base.py

@@ -5,15 +5,15 @@
 from logging import Logger, getLogger
 from typing import Any, Mapping, Type, TypeVar
 
+from bonsai_libs.parse.io.delimited import read_delimited, validate_fields
+from bonsai_libs.parse.io.types import DelimiterRow, StreamOrPath
 from bonsai_libs.parse.core.envelope import (
     default_empty_predicate,
     envelope_absent,
     envelope_skipped,
     run_as_envelope,
 )
 from bonsai_libs.parse.exceptions import UnsupportedAnalysisTypeError
-from bonsai_libs.parse.io.delimited import read_delimited, validate_fields
-from bonsai_libs.parse.io.types import DelimiterRow, StreamOrPath
 from bonsai_libs.parse.models.enums import AnalysisType, ResultStatus
 
 T = TypeVar("T")
@@ -91,7 +91,9 @@ def parse(
         out.results.update(results)
         return out
 
-    def _normalize_want(self, want: set[AnalysisType] | AnalysisType | None) -> set[AnalysisType]:
+    def _normalize_want(
+        self, want: set[AnalysisType] | AnalysisType | None
+    ) -> set[AnalysisType]:
         """Normalize the want parameter to a set of AnalysisType."""
 
         want = want or set(self.produces)
@@ -174,11 +176,13 @@ def _get_first_normalized_row(
     ) -> Mapping[str, Any] | None:
         """Convenience: read, validate and normalize a single delimited row.
 
-        ``column_map`` is passed through to :func:`bonsai_libs.io.delimited.normalize_row`.
+        ``column_map`` is passed through to :func:`prp.io.delimited.normalize_row`.
         If the source is empty this returns ``None``.  Extra rows are consumed up
         to ``max_consume`` and a warning emitted via :meth:`log_warning`.
         """
-        first, rows = self._read_rows(source, required=required, strict_columns=strict_columns)
+        first, rows = self._read_rows(
+            source, required=required, strict_columns=strict_columns
+        )
         if first is None:
             return None
         # normalization is a very common pattern; import lazily to avoid a
@@ -213,10 +217,18 @@ def _parse_impl(
 class SingleAnalysisParser(BaseParser):
     """Abtracted parser class for softwares that produces exactly one AnalysisType"""
 
+    subcommand: str | None = None
+    """Optional subcommand identifier used when one software binary produces
+    multiple distinct output formats (e.g. 'coverage' and 'stats' for samtools).
+    The registry uses (software, subcommand) as a composite key so each output
+    type can have its own parser class."""
+
     def __init_subclass__(cls, **kwargs):
         super().__init_subclass__(**kwargs)
         if not hasattr(cls, "produces") or len(cls.produces) != 1:
-            raise TypeError(f"{cls.__name__}.produces must contain exactly one AnalysisType")
+            raise TypeError(
+                f"{cls.__name__}.produces must contain exactly one AnalysisType"
+            )
 
     @property
     def analysis_type(self) -> AnalysisType:

src/bonsai_libs/parse/core/envelope.py

@@ -31,21 +31,22 @@ def envelope_from_value(
 
     status = ResultStatus.EMPTY if empty_predicate(value) else ResultStatus.PARSED
     from bonsai_libs.parse.models.base import ResultEnvelope
-
     return ResultEnvelope(status=status, value=value, reason=reason, meta=meta or {})
 
 
-def envelope_error(reason: str, *, meta: dict[str, Any] | None = None) -> "ResultEnvelope":
+def envelope_error(
+    reason: str, *, meta: dict[str, Any] | None = None
+) -> "ResultEnvelope":
     """Create an envelope that signifies that an error occured."""
     from bonsai_libs.parse.models.base import ResultEnvelope
-
     return ResultEnvelope(status=ResultStatus.ERROR, reason=reason, meta=meta or {})
 
 
-def envelope_absent(reason: str, *, meta: dict[str, Any] | None = None) -> "ResultEnvelope":
+def envelope_absent(
+    reason: str, *, meta: dict[str, Any] | None = None
+) -> "ResultEnvelope":
     """Create an envelope that for result being absent in the input file."""
     from bonsai_libs.parse.models.base import ResultEnvelope
-
     return ResultEnvelope(status=ResultStatus.ABSENT, reason=reason, meta=meta or {})
 
 
@@ -54,7 +55,6 @@ def envelope_skipped(
 ) -> "ResultEnvelope":
     """Create an envelope that signifies that the result was skipped by the user."""
     from bonsai_libs.parse.models.base import ResultEnvelope
-
     return ResultEnvelope(status=ResultStatus.SKIPPED, reason=reason, meta=meta or {})
 
 
@@ -75,7 +75,9 @@ def run_as_envelope(
         value = fn()
     except AbsentResultError as exc:
         if logger:
-            logger.info("result absent", extra={"context": {**base_meta, "error": str(exc)}})
+            logger.info(
+                "result absent", extra={"context": {**base_meta, "error": str(exc)}}
+            )
         return envelope_absent(reason=reason_if_absent or str(exc), meta=base_meta)
 
     except ParserError as exc:
@@ -93,7 +95,9 @@ def run_as_envelope(
 
     except Exception as exc:
         if logger:
-            logger.exception("Step failed", extra={"context": {**base_meta, "error": str(exc)}})
+            logger.exception(
+                "Step failed", extra={"context": {**base_meta, "error": str(exc)}}
+            )
         err_meta = {**base_meta, "exception": type(exc).__name__}
         return envelope_error(reason=str(exc), meta=err_meta)
 
@@ -112,5 +116,4 @@ def run_as_envelope(
         reason = reason_if_empty or None
 
     from bonsai_libs.parse.models.base import ResultEnvelope
-
     return ResultEnvelope(status=status, value=value, reason=reason, meta=base_meta)

src/bonsai_libs/parse/core/registry.py

@@ -47,7 +47,24 @@ def _normalize_version(version: str | Version) -> Version:
     raise TypeError(f"Version must be str or Version, got {type(version).__name__}")
 
 
-def register_parser(software: str, min_version: str | None = None, max_version: str | None = None):
+def _registry_key(software: str, subcommand: str | None) -> str:
+    """Build the registry lookup key from software and optional subcommand.
+
+    When a subcommand is given the key becomes 'software.subcommand'
+    (e.g. 'samtools.coverage', 'samtools.stats'), forming a single composite
+    key — not two separate registrations — so multiple parsers can share the
+    same software name without colliding.
+    """
+    return f"{software}.{subcommand}" if subcommand else software
+
+
+def register_parser(
+    software: str,
+    min_version: str | None = None,
+    max_version: str | None = None,
+    *,
+    subcommand: str | None = None,
+):
     """Decorator to register a parser for a range of versions.
 
     Null values means either undefined or no upper range.
@@ -61,15 +78,17 @@ def wrapper(cls: ParserRegistryEntry):
         new_min = _normalize_version(min_version)
         new_max = _normalize_version(max_version)
 
+        key = _registry_key(software, subcommand)
+
         # Fetch existing ranges for this software
-        existing_ranges = _PARSER_REGISTRY.get(software, [])
+        existing_ranges = _PARSER_REGISTRY.get(key, [])
 
         # Check for overlapping version ranges
         for span in existing_ranges:
             if not (new_max < span.min_version or new_min > span.max_version):
                 # Ranges overlap → safety error
                 raise ValueError(
-                    f"Cannot register parser {cls.__name__} for software '{software}' "
+                    f"Cannot register parser {cls.__name__} for software '{key}' "
                     f"with version range [{new_min}, {new_max}] because it overlaps "
                     f"with existing parser {span.entry.__name__} range "
                     f"[{span.min_version}, {span.max_version}]."
@@ -81,30 +100,36 @@ def wrapper(cls: ParserRegistryEntry):
             max_version=new_max,
             entry=cls,
         )
-        _PARSER_REGISTRY.setdefault(software, []).append(v_range)
+        _PARSER_REGISTRY.setdefault(key, []).append(v_range)
 
         return cls
 
     return wrapper
 
 
-def get_parser(software: str, *, version: str) -> ParserRegistryEntry:
+def get_parser(
+    software: str, *, version: str, subcommand: str | None = None
+) -> ParserRegistryEntry:
     """Get parser from registry."""
     if not isinstance(software, str):
         raise TypeError(f"`software` must be str, got {type(software).__name__}")
 
-    if software not in registered_softwares():
-        raise UnsupportedSoftwareError(f"No parser registered for software: {software}")
+    key = _registry_key(software, subcommand)
+
+    if key not in registered_softwares():
+        raise UnsupportedSoftwareError(f"No parser registered for software: {key}")
 
     # Normalize version to PkgVersion
     v = _normalize_version(version)
 
-    for span in sorted(_PARSER_REGISTRY[software], key=lambda r: (r.min_version, r.max_version)):
+    for span in sorted(_PARSER_REGISTRY[key], key=lambda r: (r.min_version, r.max_version)):
         if span.min_version <= v <= span.max_version:
             return span.entry
 
     # Return the correct error.
-    raise UnsupportedVersionError(f"No parser available for software '{software}' version {v}")
+    raise UnsupportedVersionError(
+        f"No parser available for software '{key}' version {v}"
+    )
 
 
 def registered_softwares() -> list[str]:
@@ -137,6 +162,7 @@ def run_parser(
     *,
     version: str,
     data: StreamOrPath,
+    subcommand: str | None = None,
     want: set[AnalysisType] | None = None,
     parser_init: dict[str, Any] | None = None,
     **parse_kwargs: Any,
@@ -146,7 +172,7 @@ def run_parser(
     if not isinstance(software, (AnalysisSoftware, str)):
         raise ValueError(f"Invalid input for 'run_parser', got {type(software)}")
 
-    entry = get_parser(software, version=version)
+    entry = get_parser(software, version=version, subcommand=subcommand)
     parse_fn = resolve_parser(entry, **(parser_init or {}))
     ev = parse_fn(data, want=want, **parse_kwargs)
     # add version to results
@@ -233,9 +259,7 @@ def _hydrate_raw_value(model_cls: ModelClass | TypeAdapter, raw_value: Any) -> A
     raise TypeError(f"Unsupported nested element model type: {type(model_cls).__name__}")
 
 
-def _hydrate_nested_fields(
-    result_obj: BaseModel, field_models: dict[str, ModelClass | TypeAdapter]
-) -> BaseModel:
+def _hydrate_nested_fields(result_obj: BaseModel, field_models: dict[str, ModelClass | TypeAdapter]) -> BaseModel:
     updates: dict[str, Any] = {}
 
     for field_name, model_cls in field_models.items():

src/bonsai_libs/parse/exceptions.py

@@ -1,4 +1,4 @@
-"""Common errors for bonsai_libs.parse."""
+"""Common errors"""
 
 from typing import Any
 
@@ -11,6 +11,10 @@ def __init__(self, message: str, *, context: dict[str, Any] | None = None):
         self.context = context or {}
 
 
+class DataFormatError(ParserError):
+    """Generic data/serialization errors (IO-level)."""
+
+
 class InvalidDataFormat(ParserError):
     """Fatal: content present but corrupted/ill-formed -> ERROR."""
 

src/bonsai_libs/parse/io/delimited.py

@@ -127,7 +127,9 @@ def validate_fields(
     extra = (cols - allowed) if strict else set()
 
     if missing:
-        raise ValueError(f"Missing required columns: {sorted(missing)}; got: {sorted(cols)}")
+        raise ValueError(
+            f"Missing required columns: {sorted(missing)}; got: {sorted(cols)}"
+        )
     if strict and extra:
         raise ValueError(f"Unexpected extra columns: {sorted(extra)}")
 

src/bonsai_libs/parse/io/json.py

@@ -3,7 +3,7 @@
 import json
 from typing import Any, Mapping
 
-from bonsai_libs.parse.exceptions import InvalidDataFormat
+from bonsai_libs.parse.exceptions import DataFormatError
 
 from .types import StreamOrPath
 from .utils import ensure_text_stream
@@ -19,7 +19,9 @@ def read_json(source: StreamOrPath, *, encoding: str = "utf-8") -> Any:
         stream = ensure_text_stream(source, encoding=encoding)
         return json.loads(stream.read())
     except TypeError as exc:
-        raise DataFormatError(f"Failed to read JSON from source of type {type(source)!r}") from exc
+        raise DataFormatError(
+            f"Failed to read JSON from source of type {type(source)!r}"
+        ) from exc
 
 
 def require_mapping(obj: Any, *, what: str) -> Mapping[str, Any]:

src/bonsai_libs/parse/io/utils.py

@@ -66,7 +66,7 @@ def convert_rel_to_abs_path(path: str, validation_info: ValidationInfo) -> Path:
     """Validate that file exist and resolve realtive directories.
 
     if a path is relative, convert to absolute from the configs parent directory
-    i.e.  path = ./results/sample_name.json --> /path/to/sample_name.json
+    i.e.  prp_path = ./results/sample_name.json --> /path/to/sample_name.json
           given, cnf_path = /data/samples/cnf.yml
     relative paths are used when bootstraping a test database
     """