Add Rimsky orchestration bridge for RIFT

.github/workflows/rimsky-integration.yml

@@ -0,0 +1,54 @@
+name: Rimsky integration
+
+on:
+  push:
+    branches: [rift_O4c]
+  pull_request:
+    branches: [rift_O4c]
+  workflow_dispatch:
+
+jobs:
+  end-to-end:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - rimsky-series: '0.1.0rc1'
+            rimsky-spec: 'rimsky==0.1.0rc1'
+            python-version: '3.12'
+          - rimsky-series: 'main-2621d15'
+            rimsky-spec: 'git+https://git.ligo.org/colm.talbot/rimsky.git@2621d15cf9a39ce01145ac5b81c92e1173a1d2d0'
+            python-version: '3.14'
+    name: end-to-end (${{ matrix.rimsky-series }})
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          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 RIFT and Rimsky orchestration stack
+        run: |
+          retry_pip_install() {
+            local attempt
+            for attempt in 1 2 3; do
+              if python -m pip install "$@" --break-system-packages; then
+                return 0
+              fi
+              echo "pip install failed (attempt ${attempt}/3)"
+              sleep 10
+            done
+            return 1
+          }
+          python -m pip install --upgrade pip --break-system-packages
+          retry_pip_install '${{ matrix.rimsky-spec }}'
+          retry_pip_install --upgrade 'asimov>=0.7,<0.8' 'bilby_pipe @ git+https://git.ligo.org/lscsoft/bilby_pipe.git@be6c77021db809690c781a7744f40564e62dd72f'
+          python -m pip install pytest packaging --break-system-packages
+          python -m pip install --editable . --no-deps --break-system-packages
+      - name: Run Rimsky to RIFT end-to-end test
+        run: python -m pytest -q MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py

.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+*.py[cod]
+.pytest_cache/
+asimov.log

MonteCarloMarginalizeCode/Code/RIFT/asimov/README.md

@@ -20,3 +20,12 @@ finished and does not submit a duplicate postprocessing job.
 contract for separate postprocessing adapters: samples (always a list), the
 RIFT configuration, PSDs, calibration envelopes, likelihood products, and
 basic event/analysis provenance.  Consumers should tolerate additional keys.
+
+Rimsky integration
+------------------
+
+The ``rift-rimsky-analysis`` command generates a RIFT follow-up document for
+Rimsky's ``sample_sink.asimov_configuration`` hook. It bootstraps from the
+PESummary metafile produced by Rimsky's online Bilby analysis and normalizes
+Rimsky's underscore-separated prior names for the RIFT template. See
+``RIFT/rimsky/README.md`` for configuration and operational details.

MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini

@@ -62,12 +62,17 @@ accounting_group_user={{ config['condor']['user'] }}
 
 [datafind]
 url-type=file
+{% if data contains 'frame types' %}
 types = { {% for ifo in ifos %}"{{ifo}}":"{{data['frame types'][ifo]}}",{% endfor %} }
+{% endif %}
 
 [data]
 channels = { {% for ifo in ifos %}"{{ifo}}":"{{data['channels'][ifo]}}",{% endfor %} }
 
 [lalinference]
+{% if data contains 'frame cache' %}
+fake-cache = { {% for ifo in ifos %}"{{ifo}}":"{{data['frame cache'][ifo]}}",{% endfor %} }
+{% endif %}
 {% if likelihood contains 'minimum frequency' %}
 flow = { {% for ifo in ifos %}"{{ifo}}":{{likelihood['minimum frequency'][ifo]}},{% endfor %} }
 {% else %}
@@ -293,11 +298,12 @@ l-max={{ waveform['maximum mode'] | default: 4 }}
 # * distance prior if this argument is *not* set is dL^2
 {%- if priors.keys() contains "luminosity distance" %}
 {%- assign p = priors['luminosity distance'] %}
-{% if p['type'] contains 'PowerLaw' %}
+{%- assign distance_prior_type = p['type'] | default: '' %}
+{% if distance_prior_type contains 'PowerLaw' %}
 # Default distance prior no text here, assume alpha=2
-{% elsif p['type'] contains 'UniformSourceFrame' %}
+{% elsif distance_prior_type contains 'UniformSourceFrame' %}
 ile-distance-prior='cosmo_sourceframe'
-{% elsif p['type'] contains 'UniformComovingVolume' %}
+{% elsif distance_prior_type contains 'UniformComovingVolume' %}
 ile-distance-prior='cosmo'
 {% else %}
 ile-distance-prior="pseudo_cosmo"

MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py

@@ -5,6 +5,7 @@
 import os
 import re
 import subprocess
+from pathlib import Path
 
 from ligo.gracedb.rest import HTTPError
 
@@ -72,6 +73,12 @@ def __init__(self, production, category=None):
 
     def _create_ledger_entries(self):
         """Create entries in the ledger which might be required in the templating."""
+        # Rimsky writes Bilby-style prior names into the shared Asimov event.
+        # Add RIFT's legacy aliases without removing the keys Bilby consumes.
+        from RIFT.rimsky import normalize_event_metadata
+        normalized = normalize_event_metadata(self.production.meta)
+        self.production.meta.clear()
+        self.production.meta.update(normalized)
         if "sampler" not in self.production.meta:
             self.production.meta["sampler"] = {}
         required_args = {
@@ -95,6 +102,120 @@ def _get_psds(self, format="ascii"):
         if format == "xml" and isinstance(assets, dict):
             return list(assets.values())
         return assets
+
+    def _detector_for_psd(self, psdfile):
+        """Identify a PSD's detector without assuming a filename ordering."""
+        filename = Path(psdfile).name.upper()
+        matches = [
+            ifo.upper()
+            for ifo in self.production.meta.get("interferometers", [])
+            if ifo.upper() in filename
+        ]
+        if len(matches) != 1:
+            raise PipelineException(
+                "RIFT cannot identify a unique detector for PSD {}".format(psdfile),
+                production=self.production.name,
+            )
+        return matches[0]
+
+    def _convert_psd(self, ascii_format, ifo, dryrun=False):
+        """Convert one on-disk ASCII PSD into RIFT's XML representation."""
+        ascii_format = os.path.abspath(os.path.expanduser(ascii_format))
+        if not os.path.isfile(ascii_format):
+            raise PipelineException(
+                "RIFT PSD for {} does not exist: {}".format(ifo, ascii_format),
+                production=self.production.name,
+            )
+
+        command = [
+            "convert_psd_ascii2xml",
+            "--fname-psd-ascii",
+            ascii_format,
+            "--ifo",
+            ifo.upper(),
+            "--conventional-postfix",
+        ]
+        if dryrun:
+            print(" ".join(command))
+            return command
+
+        try:
+            converted = subprocess.run(
+                command,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                check=False,
+            )
+        except FileNotFoundError as exc:
+            raise PipelineException(
+                "RIFT PSD conversion executable is unavailable: {}".format(
+                    command[0]
+                ),
+                production=self.production.name,
+            ) from exc
+        if converted.returncode != 0:
+            output = converted.stdout.decode(errors="replace")
+            raise PipelineException(
+                "RIFT could not convert the {} PSD {}:\n{}".format(
+                    ifo, ascii_format, output
+                ),
+                production=self.production.name,
+            )
+
+        xml_path = os.path.abspath("{}-psd.xml.gz".format(ifo.upper()))
+        if not os.path.isfile(xml_path):
+            raise PipelineException(
+                "RIFT PSD conversion did not create {}".format(xml_path),
+                production=self.production.name,
+            )
+        return xml_path
+
+    def _prepare_frame_caches(self):
+        """Create LAL cache files for local frames supplied by Rimsky."""
+        if self.production.meta.get("orchestrator") != "rimsky":
+            return {}
+        data = self.production.meta.get("data", {})
+        data_files = data.get("data files", {})
+        if not isinstance(data_files, dict) or not data_files:
+            return {}
+
+        cache_dir = Path(self.production.event.work_dir)
+        cache_dir.mkdir(parents=True, exist_ok=True)
+        caches = {}
+        for detector, files in data_files.items():
+            if isinstance(files, (str, os.PathLike)):
+                files = [files]
+            if not isinstance(files, (list, tuple)):
+                raise PipelineException(
+                    "RIFT Rimsky frame list for {} is malformed".format(detector),
+                    production=self.production.name,
+                )
+
+            entries = []
+            for filename in files:
+                frame = Path(filename).expanduser().resolve()
+                match = re.search(r"-(\d+)-(\d+)\.gwf$", frame.name)
+                if not frame.is_file() or match is None:
+                    raise PipelineException(
+                        "RIFT Rimsky frame is missing or has no GPS/duration suffix: {}".format(
+                            frame
+                        ),
+                        production=self.production.name,
+                    )
+                start, duration = match.groups()
+                entries.append(
+                    "{} RIMSKY {} {} {}".format(
+                        detector[0].upper(), start, duration, frame.as_uri()
+                    )
+                )
+
+            cache = cache_dir / "{}-rimsky.cache".format(detector)
+            cache.write_text("\n".join(entries) + "\n", encoding="utf-8")
+            caches[detector] = str(cache)
+
+        data["frame cache"] = caches
+        return caches
+
     # Top-level groups a PESummary metafile carries that are not analysis labels
     _PESUMMARY_RESERVED = ('version', 'history')
 
@@ -298,29 +419,42 @@ def before_config(self, dryrun=False):
         """
         event = self.production.event
         category = config.get("general", "calibration_directory")
+        self._prepare_frame_caches()
         # XML PSDs
         self.logger.info("Checking for XML format PSDs")
         if len(self._get_psds("xml")) == 0 and "psds" in self.production.meta:
             self.logger.info("Did not find XML format PSDs")
+            project_dir = Path.cwd()
+            repository_dir = Path(event.repository.directory)
+            if not repository_dir.is_absolute():
+                repository_dir = (project_dir / repository_dir).resolve()
             for ifo in self.production.meta["interferometers"]:
                 with set_directory(f"{event.work_dir}"):
                     sample = self.production.meta["likelihood"]["sample rate"]
                     self.logger.info(f"Converting {ifo} {sample}-Hz PSD to XML")
-                    self._convert_psd(
+                    asset = self._convert_psd(
                         self.production.meta["psds"][sample][ifo], ifo, dryrun=dryrun
                     )
-                    asset = f"{ifo.upper()}-psd.xml.gz"
-                    self.logger.info(f"Conversion complete as {asset}")
-                    git_location = os.path.join(category, "psds")
-                    saveloc = os.path.join(
-                        git_location, str(sample), f"psd_{ifo}.xml.gz"
-                    )
-                    self.production.event.repository.add_file(
+                if dryrun:
+                    continue
+                self.logger.info(f"Conversion complete as {asset}")
+                git_location = os.path.join(category, "psds")
+                saveloc = os.path.join(
+                    git_location, str(sample), f"psd_{ifo}.xml.gz"
+                )
+                # EventRepo paths may be relative to the Asimov project.  Add
+                # the converted file after leaving the event work directory so
+                # it cannot be nested beneath that directory accidentally.
+                with set_directory(project_dir):
+                    event.repository.add_file(
                         asset,
                         saveloc,
                         commit_message=f"Added the xml format PSD for {ifo}.",
                     )
-                    self.logger.info(f"Saved at {saveloc}")
+                xml_psds = getattr(self.production, "xml_psds", None)
+                if isinstance(xml_psds, dict):
+                    xml_psds[ifo] = str(repository_dir / saveloc)
+                self.logger.info(f"Saved at {saveloc}")
         # calmarg: find bilby ini file if needed
         self.logger.info(" About to check for calmarg ")
         if 'likelihood' in self.production.meta['sampler']:
@@ -672,7 +806,7 @@ def build_dag(self, user=None, dryrun=False):
                     if self.production.event.repository:
                         # with set_directory(os.path.abspath(self.production.rundir)):
                         for psdfile in self._get_psds("xml"):
-                            ifo = psdfile.split("/")[-1].split("-")[1].split(".")[0]
+                            ifo = self._detector_for_psd(psdfile)
                             os.system(f"cp {psdfile} {ifo}-psd.xml.gz")
 
                         # os.system("cat *_local.cache > local.cache")
@@ -715,7 +849,7 @@ def submit_dag(self, dryrun=False):
         """
         self.before_submit()
         for psdfile in self._get_psds("xml"):
-            ifo = psdfile.split("/")[-1].split("-")[1].split(".")[0]
+            ifo = self._detector_for_psd(psdfile)
             os.system(f"cp {psdfile} {ifo}-psd.xml.gz")
 
         command = [

MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md

@@ -0,0 +1,60 @@
+# Rimsky integration
+
+Rimsky performs online Bilby parameter estimation and can launch follow-up
+analyses through its Asimov hook.  RIFT supplies a bridge for that hook:
+
+1. `rift-rimsky-analysis rimsky.yaml rift-followup.yaml` reads the Rimsky
+   configuration and writes both a RIFT Asimov analysis document and a runnable
+   `rimsky-rift.yaml`.
+2. Initialize an Asimov 0.7 project named by `asimovdir` once, install RIFT in its
+   environment, and run `rimsky rimsky-rift.yaml`.
+
+The generated Rimsky configuration defaults `event_sink.bilby_pipe_format` to
+`full-submit`, points `sample_sink.asimov_configuration` at the generated RIFT
+analysis, and makes relative output paths absolute.  Thus the first online
+Bilby result is written as a PESummary metafile and Rimsky immediately adds the
+ready RIFT follow-up to Asimov.  A running Asimov manager then builds and submits
+that production.  Existing explicit Bilby run modes and Asimov project paths
+are preserved.  Use `--configured-rimsky PATH` to choose a different filename.
+
+For example:
+
+```yaml
+output_dir: ./output
+asimovdir: ./asimov
+detectors: [H1, L1, V1]
+
+sample_sink:
+  # Written automatically in rimsky-rift.yaml.
+  asimov_configuration: /absolute/path/to/rift-followup.yaml
+
+# Optional. Rimsky ignores this extra section; the RIFT generator consumes it.
+rift:
+  name: rift-online
+  waveform:
+    approximant: IMRPhenomXPHM
+  scheduler:
+    accounting group: ligo.dev.o4.cbc.pe.rift
+    osg: false
+```
+
+Rimsky writes a PESummary metafile before applying the follow-up file.  The
+generated analysis uses an absolute `output_dir/*/*/{event}/...` glob to find
+that event's metafile, sets its dataset to `bilby-online`, and bootstraps RIFT
+and its coincidence XML from the online posterior. Exactly one metafile must match; RIFT fails closed
+if the path is missing or ambiguous.
+
+Rimsky 0.1 event documents use Bilby-style prior names (`chirp_mass`,
+`mass_ratio`, `a_1`, and so on).  The RIFT pipeline retains those keys and adds
+the space-separated aliases expected by its Asimov template.  This makes the
+same event usable by both Bilby and RIFT analyses.
+
+The bridge itself consumes plain YAML mappings and does not import Rimsky.  Its
+unit tests remain isolated from streaming, GraceDB, and HTCondor.  Dedicated
+end-to-end lanes install Rimsky `0.1.0rc1` on Python 3.12 and pinned current main
+on Python 3.14, both forced onto Asimov 0.7 and the merged bilby_pipe 0.7 adapter.
+They load the generated configuration through Rimsky, invoke its real post-PE
+Asimov hook, discover the RIFT pipeline, and resolve the first metafile as the
+bootstrap input. External scheduler submission is the only mocked boundary.
+The current-main pin is commit `2621d15` (2026-09-01); the bilby_pipe adapter pin
+is `be6c770` pending its next release.