Pipeline: reformat to ascii ; asimov tests; developer guide; cip no-explode handoff

.github/workflows/ci.yml

@@ -105,6 +105,38 @@ jobs:
       - name: Run simulation_manager smoke test
         run: bash .travis/test-simulation-manager.sh
 
+  asimov-integration:
+    needs: install
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - asimov-series: '0.5'
+            asimov-spec: 'asimov>=0.5,<0.6'
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+          cache: 'pip'
+          cache-dependency-path: requirements.txt
+      - name: Enable symlink
+        run: sudo ln -sf $(which python3) /usr/bin/python
+      - name: Install system dependencies
+        run: sudo apt-get update && sudo apt-get install -y libgsl-dev
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip --break-system-packages
+          python -m pip install -r requirements.txt --break-system-packages
+          python -m pip install coverage pytest --break-system-packages
+          python -m pip install --editable . --break-system-packages
+          python -m pip install htcondor --only-binary=:all: --break-system-packages
+      - name: Install Asimov
+        run: python -m pip install '${{ matrix.asimov-spec }}' 'asimov-gwdata>=0.4,<0.5' --break-system-packages
+      - name: Run Asimov integration test
+        run: bash .travis/test-asimov.sh
+
   test-run:
     needs: install
     runs-on: ubuntu-latest
@@ -171,4 +203,4 @@ jobs:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           publish_dir: ./docs/build/
           publish_branch: gh-pages
-          force_push: true
+          force_push: true

.travis/test-asimov.sh

@@ -0,0 +1,8 @@
+#! /bin/bash
+set -euo pipefail
+
+# The RIFT Asimov plugin is currently developed and validated against the
+# Asimov 0.5 series.  This test skips cleanly for unsupported/future series
+# from inside pytest, so developers can preflight 0.6/0.7 environments without
+# editing the test.
+python -m pytest -q MonteCarloMarginalizeCode/Code/test/asimov_integration

MonteCarloMarginalizeCode/Code/RIFT/misc/__init__.py

@@ -1 +1 @@
-__all__ = ['dag_utils','xmlutils','bounded_kde','samples_utils']
+__all__ = ['dag_utils','xmlutils','bounded_kde','samples_utils','hyperpipeline_io']

MonteCarloMarginalizeCode/Code/RIFT/misc/dag_utils.py

@@ -711,6 +711,7 @@ def write_CIP_sub(tag='integrate', exe=None, input_net='all.net',output='output-
     return ile_job, ile_sub_name
 
 
+
 def write_puff_sub(tag='puffball', exe=None, base=None,input_net='output-ILE-samples',output='puffball',universe="vanilla",out_dir=None,log_dir=None, use_eos=False,ncopies=1,arg_str=None,request_memory=1024,arg_vals=None, no_grid=False,extra_text='',**kwargs):
     """
     Perform puffball calculation 
@@ -1293,7 +1294,15 @@ def write_unify_sub_simple(tag='unify', exe=None, base=None,target=None,universe
 
     """
 
-    exe = exe or which("util_CleanILE.py")  # like cat, but properly accounts for *independent* duplicates. (Danger if identical). Also strips large errors
+    if exe is None:
+        if str(os.environ.get("RIFT_HYPERPIPELINE_FORMAT", "")).strip().lower() in ("1", "true", "yes", "on"):
+            exe = which("util_CleanILE_hyperpipeline.py")
+            if not exe:
+                exe = os.path.abspath(os.path.join(
+                    os.path.dirname(__file__), "..", "..", "bin",
+                    "util_CleanILE_hyperpipeline.py"))
+        else:
+            exe = which("util_CleanILE.py")  # like cat, but properly accounts for *independent* duplicates. (Danger if identical). Also strips large errors
 
     # Write unify.sh
     #    - problem of globbing inside condor commands
@@ -2159,7 +2168,24 @@ def write_joingrids_sub(tag='join_grids', exe=None, universe='vanilla', input_pa
 
     working_dir = log_dir.replace("/logs", '') # assumption about workflow/naming! Danger!
 
-    fname_out =target_dir + "/" +output_base + ".xml.gz"
+    # Hyperpipeline mode: emit a .dat composite via the EOS-style
+    # head-line + cat | sort | uniq | shuf shell pattern (mirrors
+    # join_post.sh in create_eos_posterior_pipeline).  XML helpers
+    # (igwn_ligolw_add) are not used.
+    import os as _os_join
+    _hpip_join = str(_os_join.environ.get("RIFT_HYPERPIPELINE_FORMAT", "")).strip().lower() in ("1","true","yes","on")
+    _suffix_join = "dat" if _hpip_join else "xml.gz"
+    _shuffle_filter_join = None
+    if _hpip_join:
+        _shuffle_filter_join = which("shuf") or which("gshuf")
+        if _shuffle_filter_join is None:
+            _sort_join = which("sort")
+            if _sort_join is not None:
+                _shuffle_filter_join = _sort_join + " -R"
+            else:
+                _shuffle_filter_join = "cat"
+
+    fname_out =target_dir + "/" +output_base + "." + _suffix_join
     if n_explode ==1:   # we are really doing a glob match
         fname_out = fname_out.replace('$(macroiteration)','$1')
         fname_out = fname_out.replace('$(macroiterationnext)','$2')
@@ -2169,11 +2195,34 @@ def write_joingrids_sub(tag='join_grids', exe=None, universe='vanilla', input_pa
         if old_add:
             extra_arg = " --ilwdchar-compat "  # should never be used anymore
         with open("join_grids.sh",'w') as f:
-            f.write("#! /bin/bash  \n")
-            f.write(r"""
+            if _hpip_join:
+                # Pick any one shard to source the column header from, then
+                # concatenate every shard's data rows (skipping `#` lines)
+                # and shuffle so spokes are interleaved.
+                f.write("#! /bin/bash  \n")
+                f.write(r"""
+# merge hyperpipeline ASCII shards
+{extra}
+set -e
+shopt -s nullglob
+SHARDS=({work}/{out}*.{suf})
+if [ ${{#SHARDS[@]}} -eq 0 ]; then
+  echo "join_grids.sh: no input shards matched {work}/{out}*.{suf}" >&2
+  exit 1
+fi
+# Header: take the first comment block from the first shard.
+grep -E '^#' "${{SHARDS[0]}}" > {out_path}
+# Body: every non-comment line from every shard, deduped + shuffled.
+grep -hE -v '^#' "${{SHARDS[@]}}" | sort -u | {shuffle_filter} >> {out_path}
+""".format(extra=extra_text, work=alt_work_dir, out=alt_out,
+           suf=_suffix_join, out_path=fname_out,
+           shuffle_filter=_shuffle_filter_join))
+            else:
+                f.write("#! /bin/bash  \n")
+                f.write(r"""
 # merge using glob command called from shell
 {}
-{}  {} --output {}  {}/{}*.xml.gz 
+{}  {} --output {}  {}/{}*.xml.gz
 """.format(extra_text,exe,extra_arg,fname_out,alt_work_dir,alt_out))
         os.system("chmod a+x join_grids.sh")
         exe = target_dir + "/join_grids.sh"
@@ -3066,4 +3115,3 @@ def write_hyperpost_sub(tag='HYPER', exe=None, input_net='all.marg_net',output='
 
 
     return ile_job, ile_sub_name
-

MonteCarloMarginalizeCode/Code/RIFT/misc/dag_utils_generic.py

@@ -2666,7 +2666,15 @@ def write_unify_sub_simple(tag='unify', exe=None, base=None,target=None,universe
 
     """
 
-    exe = exe or which("util_CleanILE.py")  # like cat, but properly accounts for *independent* duplicates. (Danger if identical). Also strips large errors
+    if exe is None:
+        if str(os.environ.get("RIFT_HYPERPIPELINE_FORMAT", "")).strip().lower() in ("1", "true", "yes", "on"):
+            exe = which("util_CleanILE_hyperpipeline.py")
+            if not exe:
+                exe = os.path.abspath(os.path.join(
+                    os.path.dirname(__file__), "..", "..", "bin",
+                    "util_CleanILE_hyperpipeline.py"))
+        else:
+            exe = which("util_CleanILE.py")  # like cat, but properly accounts for *independent* duplicates. (Danger if identical). Also strips large errors
 
     # Write unify.sh
     #    - problem of globbing inside condor commands
@@ -3532,7 +3540,22 @@ def write_joingrids_sub(tag='join_grids', exe=None, universe='vanilla', input_pa
 
     working_dir = log_dir.replace("/logs", '') # assumption about workflow/naming! Danger!
 
-    fname_out =target_dir + "/" +output_base + ".xml.gz"
+    # Hyperpipeline mode: emit a .dat composite via the EOS-style
+    # head-line + cat | sort | uniq | shuf shell pattern.  XML helpers
+    # (igwn_ligolw_add) are not used.
+    _hpip_join = str(os.environ.get("RIFT_HYPERPIPELINE_FORMAT", "")).strip().lower() in ("1","true","yes","on")
+    _suffix_join = "dat" if _hpip_join else "xml.gz"
+    _shuffle_filter_join = None
+    if _hpip_join:
+        _shuffle_filter_join = which("shuf") or which("gshuf")
+        if _shuffle_filter_join is None:
+            _sort_join = which("sort")
+            if _sort_join is not None:
+                _shuffle_filter_join = _sort_join + " -R"
+            else:
+                _shuffle_filter_join = "cat"
+
+    fname_out =target_dir + "/" +output_base + "." + _suffix_join
     if n_explode ==1:   # we are really doing a glob match
         fname_out = fname_out.replace('$(macroiteration)','$1')
         fname_out = fname_out.replace('$(macroiterationnext)','$2')
@@ -3542,8 +3565,28 @@ def write_joingrids_sub(tag='join_grids', exe=None, universe='vanilla', input_pa
         if old_add:
             extra_arg = " --ilwdchar-compat "  # should never be used anymore
         with open("join_grids.sh",'w') as f:
-            f.write("#! /bin/bash  \n")
-            f.write(r"""
+            if _hpip_join:
+                f.write("#! /bin/bash  \n")
+                f.write(r"""
+# merge hyperpipeline ASCII shards
+{extra}
+set -e
+shopt -s nullglob
+SHARDS=({work}/{out}*.{suf})
+if [ ${{#SHARDS[@]}} -eq 0 ]; then
+  echo "join_grids.sh: no input shards matched {work}/{out}*.{suf}" >&2
+  exit 1
+fi
+# Header: take the first comment block from the first shard.
+grep -E '^#' "${{SHARDS[0]}}" > {out_path}
+# Body: every non-comment line from every shard, deduped + shuffled.
+grep -hE -v '^#' "${{SHARDS[@]}}" | sort -u | {shuffle_filter} >> {out_path}
+""".format(extra=extra_text, work=alt_work_dir, out=alt_out,
+           suf=_suffix_join, out_path=fname_out,
+           shuffle_filter=_shuffle_filter_join))
+            else:
+                f.write("#! /bin/bash  \n")
+                f.write(r"""
 # merge using glob command called from shell
 {}
 {}  {} --output {}  {}/{}*.xml.gz 
@@ -3584,10 +3627,10 @@ def write_joingrids_sub(tag='join_grids', exe=None, universe='vanilla', input_pa
 #    ile_job.add_condor_cmd("MY.PostCmd",  ' "' + gzip + ' ' +fname_out + '"')
 
     explode_str = ""
-    explode_str += " {}/{}.xml.gz ".format(working_dir,output_base)  # base result from fitting job
+    explode_str += " {}/{}.{} ".format(working_dir,output_base,_suffix_join)  # base result from fitting job
     if n_explode >1:
      for indx in np.arange(n_explode):
-        explode_str+= " {}/{}-{}.xml.gz ".format(working_dir,output_base,indx)
+        explode_str+= " {}/{}-{}.{} ".format(working_dir,output_base,indx,_suffix_join)
         ile_job.add_arg(explode_str)
     else:
         ile_job.add_arg(" $(macroiteration) $(macroiterationnext) ")
@@ -4437,4 +4480,3 @@ def write_hyperpost_sub(tag='HYPER', exe=None, input_net='all.marg_net',output='
 
 
     return ile_job, ile_sub_name
-