v1.0.3 return non-hits to user in result dir

.gitignore

@@ -23,3 +23,4 @@ tmp/
 # Allow benchmark time/resource stats for manuscript
 # benchmark results removed, users can recapitulate easily using py script in dir [file sizes]
 !benchmarks/performance/results
+test_config.json

docs/conf.py

@@ -11,7 +11,7 @@
 author = 'Barry Digby'
 
 release = '1.0'
-version = '1.0.2'
+version = '1.0.3'
 
 # -- General configuration
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pycircdb"
-version = "1.0.2"
+version = "1.0.3"
 description = "pycircdb: integrated circRNA database annotation for computational workflows."
 readme = "README.md"
 authors = [{ name = "Barry Digby", email = "b.digby237@gmail.com" }]

utils/detect_inputs/detect_inputs_driver.py

@@ -25,7 +25,7 @@ def instantiate_driver(config: Dict[str, Any], verbose: int = 1):
     )
 
     result = dr.execute(
-        ['return_collected_results'], 
+        ['return_collected_results', 'write_no_hits'], 
         inputs={'config': config, 'lookup_tables': lookup_tables}
     )['return_collected_results']
 

utils/detect_inputs/detect_inputs_subdag.py

@@ -1,5 +1,7 @@
+import os
 import re
 import polars as pl
+from pathlib import Path
 from typing import Dict, Any, Tuple
 from polars import col
 from hamilton.htypes import Parallelizable, Collect
@@ -123,3 +125,87 @@ def return_collected_results(database_lookup: Collect[Dict[str, Dict[str, pl.Dat
                 merged_results[sample_name] = {}
             merged_results[sample_name].update(db_dict)
     return merged_results
+
+
+def _reference_hit_sets(hits: pl.DataFrame, reference: str) -> Tuple[set, set]:
+    """Build stranded/strand-stripped key sets from a database's already-filtered hits."""
+    if hits.is_empty() or reference not in hits.columns:
+        return set(), set()
+    values = [v for v in hits[reference].to_list() if v is not None]
+    stranded = set(values)
+    posonly = {re.sub(r'\|[+-]$', '', v) for v in values}
+    return stranded, posonly
+
+
+def _raw_id_matched(raw_id: str, stranded_hit_set: set, posonly_hit_set: set) -> bool:
+    """Check whether a raw input ID, with the same +/-1 tolerance as the lookup filter, hit a database."""
+    match = _COORD_RE.match(raw_id)
+    if match is None:
+        return False
+    chrom, start, end, strand = match.group(1), int(match.group(2)), match.group(3), match.group(4)
+    for shifted_start in (start, start - 1, start + 1):
+        if shifted_start < 0:
+            continue
+        pos = f"{chrom}:{shifted_start}-{end}"
+        if strand:
+            if pos + strand in stranded_hit_set:
+                return True
+        elif pos in posonly_hit_set:
+            return True
+    return False
+
+
+def _matched_raw_ids(raw_ids: list, hits_by_db: Dict[str, pl.DataFrame], reference: str) -> set:
+    """Union raw IDs that hit at least one database's lookup table for a sample."""
+    matched: set = set()
+    for hits in hits_by_db.values():
+        stranded_hit_set, posonly_hit_set = _reference_hit_sets(hits, reference)
+        if not stranded_hit_set and not posonly_hit_set:
+            continue
+        matched.update(
+            raw_id for raw_id in raw_ids
+            if raw_id not in matched and _raw_id_matched(raw_id, stranded_hit_set, posonly_hit_set)
+        )
+    return matched
+
+
+def _write_sample_no_hits(sample_name: str, sample_info: Dict[str, Any], hits_by_db: Dict[str, pl.DataFrame], output_dir: str) -> None:
+    """Write unmatched raw IDs for a single sample to no_hits.txt, if any."""
+    file_path = sample_info.get("file_path")
+    reference = sample_info.get("reference")
+    if not file_path or not reference:
+        return
+    try:
+        raw_ids = [ln.strip() for ln in Path(file_path).read_text().splitlines() if ln.strip()]
+    except Exception:
+        return
+
+    matched = _matched_raw_ids(raw_ids, hits_by_db, reference)
+    unmatched = [raw_id for raw_id in raw_ids if raw_id not in matched]
+    if not unmatched:
+        return
+
+    p = Path(output_dir)
+    if p.is_absolute():
+        output_path = p / f"{sample_name}_no_hits.txt"
+    else:
+        output_path = Path(os.getcwd()) / output_dir / f"{sample_name}_no_hits.txt"
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text("\n".join(unmatched) + "\n")
+
+
+def write_no_hits(
+    return_collected_results: Dict[str, Dict[str, pl.DataFrame]],
+    config: Dict[str, Any],
+) -> None:
+    """
+    Write input circRNAs that matched no database's lookup table to no_hits.txt.
+
+    Depends on the already-collected per-sample hits rather than running its own
+    parallel/collect pipeline, since Hamilton's dynamic execution does not support
+    requesting two independent Collect() reductions in a single driver.execute() call.
+    """
+    output_dir = config.get("global_parameters", {}).get("output_dir", "results/")
+    for sample_name, sample_info in config.get("samples", {}).items():
+        _write_sample_no_hits(sample_name, sample_info, return_collected_results.get(sample_name, {}), output_dir)