From 36420cc0710245d3b464c621c6271d48d04d530e Mon Sep 17 00:00:00 2001 From: Richard O'Shaughnessy Date: Wed, 2 Sep 2026 09:09:50 -0700 Subject: [PATCH 1/9] Add Rimsky orchestration bridge --- .gitignore | 4 + .../Code/RIFT/asimov/README.md | 9 + .../Code/RIFT/asimov/rift.ini | 3 + .../Code/RIFT/asimov/rift.py | 79 +++++- .../Code/RIFT/rimsky/README.md | 47 ++++ .../Code/RIFT/rimsky/__init__.py | 23 ++ .../Code/RIFT/rimsky/integration.py | 243 +++++++++++++++++ .../Code/test/test_rimsky_integration.py | 248 ++++++++++++++++++ setup.py | 2 + 9 files changed, 653 insertions(+), 5 deletions(-) create mode 100644 .gitignore create mode 100644 MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md create mode 100644 MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py create mode 100644 MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py create mode 100644 MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 000000000..777d51df0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +asimov.log diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/README.md b/MonteCarloMarginalizeCode/Code/RIFT/asimov/README.md index f2b805f46..d5af0c129 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/README.md +++ b/MonteCarloMarginalizeCode/Code/RIFT/asimov/README.md @@ -7,3 +7,12 @@ Based on See related documentation and examples in * https://asimov.docs.ligo.org/asimov/master/pipelines-dev.html * https://git.ligo.org/asimov/pipelines/gwdata/-/blob/master/datafind/asimov.py + +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. diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini index db21ac80d..e08c69b4f 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini +++ b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini @@ -68,6 +68,9 @@ types = { {% for ifo in ifos %}"{{ifo}}":"{{data['frame types'][ifo]}}",{% endfo 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 %} flow = { {% for ifo in ifos %}"{{ifo}}":{{quality['minimum frequency'][ifo]}},{% endfor %} } fhigh = { {% for ifo in ifos %}"{{ifo}}":{{quality['maximum frequency'][ifo]}},{% endfor %} } diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py index 1fa57bd42..ea11ae66f 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py +++ b/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 @@ -67,6 +68,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 = { @@ -78,6 +85,62 @@ def _create_ledger_entries(self): for section_arg in required_args[section]: if section_arg not in section_data: section_data[section_arg] = {} + + def _get_psds(self, format="ascii"): + """Return PSD assets across the Asimov 0.5 and 0.6 APIs.""" + legacy_getter = getattr(self.production, "get_psds", None) + if callable(legacy_getter): + assets = legacy_getter(format) + else: + attribute = "xml_psds" if format == "xml" else "psds" + assets = getattr(self.production, attribute, {}) or {} + if format == "xml" and isinstance(assets, dict): + return list(assets.values()) + return assets + + def _prepare_frame_caches(self): + """Create LAL cache files for local frames supplied by Rimsky.""" + 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') @@ -246,9 +309,10 @@ 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.production.get_psds("xml")) == 0 and "psds" in self.production.meta: + if len(self._get_psds("xml")) == 0 and "psds" in self.production.meta: self.logger.info("Did not find XML format PSDs") for ifo in self.production.meta["interferometers"]: with set_directory(f"{event.work_dir}"): @@ -268,6 +332,11 @@ def before_config(self, dryrun=False): saveloc, commit_message=f"Added the xml format PSD for {ifo}.", ) + xml_psds = getattr(self.production, "xml_psds", None) + if isinstance(xml_psds, dict): + xml_psds[ifo] = os.path.join( + self.production.event.repository.directory, saveloc + ) self.logger.info(f"Saved at {saveloc}") # calmarg: find bilby ini file if needed self.logger.info(" About to check for calmarg ") @@ -625,7 +694,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.production.get_psds("xml"): + for psdfile in self._get_psds("xml"): ifo = psdfile.split("/")[-1].split("-")[1].split(".")[0] os.system(f"cp {psdfile} {ifo}-psd.xml.gz") @@ -668,7 +737,7 @@ def submit_dag(self, dryrun=False): This will be raised if the pipeline fails to submit the job. """ self.before_submit() - for psdfile in self.production.get_psds("xml"): + for psdfile in self._get_psds("xml"): ifo = psdfile.split("/")[-1].split("-")[1].split(".")[0] os.system(f"cp {psdfile} {ifo}-psd.xml.gz") @@ -679,12 +748,12 @@ def submit_dag(self, dryrun=False): "marginalize_intrinsic_parameters_BasicIterationWorkflow.dag", ] if dryrun: - for psdfile in self.production.get_psds("xml"): + for psdfile in self._get_psds("xml"): print(f"cp {psdfile} {self.production.rundir}/{psdfile.split('/')[-1]}") print("") print(" ".join(command)) else: - for psdfile in self.production.get_psds("xml"): + for psdfile in self._get_psds("xml"): os.system( f"cp {psdfile} {self.production.rundir}/{psdfile.split('/')[-1]}" ) diff --git a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md new file mode 100644 index 000000000..0f1f4f697 --- /dev/null +++ b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md @@ -0,0 +1,47 @@ +# 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 a RIFT Asimov analysis document. +2. Set `sample_sink.asimov_configuration` in `rimsky.yaml` to the absolute path + of `rift-followup.yaml`. +3. Set `asimovdir` to an initialized Asimov project in which the RIFT package is + installed and its pipeline is configured. + +For example: + +```yaml +output_dir: ./output +asimovdir: ./asimov +detectors: [H1, L1, V1] + +sample_sink: + 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 consumes plain YAML mappings and does not import Rimsky. It is +therefore lightweight to test and isolated from Rimsky's streaming, GraceDB, +and HTCondor dependencies. The contract targets Rimsky `0.1.0rc1` and current +main as of 2026-09-02. diff --git a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py new file mode 100644 index 000000000..c75962a30 --- /dev/null +++ b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py @@ -0,0 +1,23 @@ +"""Rimsky-to-RIFT orchestration helpers. + +The public API is intentionally independent of Rimsky's Python internals. Rimsky +configuration and event documents are plain mappings, which keeps this bridge +usable across Rimsky release candidates without importing its large online-PE +runtime stack. +""" + +from .integration import ( + RimskyIntegrationError, + build_analysis, + load_rimsky_config, + normalize_event_metadata, + write_analysis, +) + +__all__ = [ + "RimskyIntegrationError", + "build_analysis", + "load_rimsky_config", + "normalize_event_metadata", + "write_analysis", +] diff --git a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py new file mode 100644 index 000000000..37c699049 --- /dev/null +++ b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py @@ -0,0 +1,243 @@ +"""Generate a RIFT follow-up for the Rimsky online-PE orchestrator.""" + +import argparse +import copy +import json +from pathlib import Path + + +class RimskyIntegrationError(ValueError): + """Raised when a Rimsky configuration cannot define a RIFT follow-up.""" + + +def _deep_update(target, updates): + """Recursively apply mapping ``updates`` without sharing mutable values.""" + for key, value in updates.items(): + if isinstance(value, dict) and isinstance(target.get(key), dict): + _deep_update(target[key], value) + else: + target[key] = copy.deepcopy(value) + return target + + +def load_rimsky_config(path): + """Load a Rimsky YAML configuration and return it as a mapping.""" + path = Path(path) + try: + import yaml + except ImportError as exc: # Rimsky itself depends on PyYAML. + raise RimskyIntegrationError( + "PyYAML is required to read a Rimsky configuration" + ) from exc + + with path.open("r", encoding="utf-8") as stream: + config = yaml.safe_load(stream) or {} + if not isinstance(config, dict): + raise RimskyIntegrationError("Rimsky configuration must be a mapping") + return config + + +def _detectors(config): + detectors = config.get("detectors", ["H1", "L1"]) + if isinstance(detectors, str): + detectors = [item.strip() for item in detectors.split(",") if item.strip()] + if not isinstance(detectors, (list, tuple)) or not detectors: + raise RimskyIntegrationError("Rimsky 'detectors' must be a non-empty list") + if not all(isinstance(detector, str) and detector for detector in detectors): + raise RimskyIntegrationError("Each Rimsky detector must be a non-empty string") + return list(detectors) + + +def _frequency_dict(value, detectors, field): + if isinstance(value, (int, float)): + return {detector: value for detector in detectors} + if not isinstance(value, dict): + raise RimskyIntegrationError( + "{} must be a number or detector mapping".format(field) + ) + missing = [detector for detector in detectors if detector not in value] + if missing: + raise RimskyIntegrationError( + "{} is missing detector(s): {}".format(field, ", ".join(missing)) + ) + selected = {detector: value[detector] for detector in detectors} + if not all(isinstance(item, (int, float)) for item in selected.values()): + raise RimskyIntegrationError("{} values must be numeric".format(field)) + return selected + + +def _resolve_output_dir(config, config_path=None): + output_dir = Path(config.get("output_dir", "output")).expanduser() + if not output_dir.is_absolute(): + base = Path(config_path).resolve().parent if config_path else Path.cwd() + output_dir = base / output_dir + return output_dir.resolve() + + +def build_analysis(config, *, config_path=None, overrides=None): + """Build the Asimov analysis document consumed by Rimsky's follow-up hook. + + Parameters + ---------- + config : mapping + Parsed Rimsky configuration. An optional top-level ``rift`` mapping is + ignored by Rimsky and accepted here as user overrides. + config_path : path-like, optional + Location of the Rimsky YAML file. Relative ``output_dir`` values are + resolved relative to this file, matching Rimsky's current behaviour. + overrides : mapping, optional + Programmatic overrides applied after the top-level ``rift`` mapping. + + Returns + ------- + dict + One Asimov analysis document suitable for + ``sample_sink.asimov_configuration``. + """ + if not isinstance(config, dict): + raise RimskyIntegrationError("Rimsky configuration must be a mapping") + + detectors = _detectors(config) + event_sink = config.get("event_sink") or {} + bilby = event_sink.get("bilby_pipe_defaults") or {} + minimum = _frequency_dict( + bilby.get("minimum_frequency", 20), detectors, "minimum_frequency" + ) + maximum = _frequency_dict( + bilby.get("maximum_frequency", 1024), detectors, "maximum_frequency" + ) + output_dir = _resolve_output_dir(config, config_path=config_path) + + # Rimsky stores each event under output_dir/YYMM/DD/SID and writes this + # metafile immediately before invoking the configured Asimov follow-ups. + bootstrap = output_dir / "*" / "*" / "{event}" / "results_page" / "metafile.hdf5" + + analysis = { + "kind": "analysis", + "name": "rift-online", + "status": "Ready", + "pipeline": "RIFT", + "comment": "RIFT follow-up launched by Rimsky after online Bilby PE", + "dataset": "bilby-online", + "likelihood": { + "start frequency": min(minimum.values()), + "minimum frequency": minimum, + "assume": {"precessing": True}, + "marginalization": {"distance": True}, + }, + "quality": { + "minimum frequency": minimum, + "maximum frequency": maximum, + }, + "waveform": { + "approximant": "IMRPhenomXPHM", + "pn amplitude order": 5, + "maximum mode": 4, + }, + "priors": { + "mass 1": {"minimum": 1, "maximum": 1000}, + }, + "sampler": {"cip": {}, "ile": {}}, + "scheduler": { + "accounting group": "ligo.dev.o4.cbc.pe.rift", + "bootstrap coinc": True, + "bootstrap file": str(bootstrap), + "osg": False, + }, + } + + configured = config.get("rift") or {} + if not isinstance(configured, dict): + raise RimskyIntegrationError( + "Optional Rimsky 'rift' settings must be a mapping" + ) + _deep_update(analysis, configured) + if overrides is not None: + if not isinstance(overrides, dict): + raise RimskyIntegrationError("RIFT overrides must be a mapping") + _deep_update(analysis, overrides) + return analysis + + +def normalize_event_metadata(metadata): + """Return RIFT-compatible metadata from a Rimsky-created event mapping. + + Rimsky 0.1 emits Bilby parameter names with underscores. RIFT's Asimov + template predates that convention and uses names containing spaces. Keep + both spellings so other analyses in the same ledger are unaffected. + """ + normalized = copy.deepcopy(metadata) + priors = normalized.setdefault("priors", {}) + aliases = { + "chirp_mass": "chirp mass", + "mass_ratio": "mass ratio", + "luminosity_distance": "luminosity distance", + "mass_1": "mass 1", + } + for source, destination in aliases.items(): + if destination not in priors and source in priors: + priors[destination] = copy.deepcopy(priors[source]) + + for source, destination in (("a_1", "spin 1"), ("a_2", "spin 2")): + if destination not in priors and source in priors: + prior = priors[source] + if isinstance(prior, dict) and "maximum" in prior: + priors[destination] = {"maximum": prior["maximum"]} + for source, destination in (("chi_1", "spin 1"), ("chi_2", "spin 2")): + if destination not in priors and source in priors: + priors[destination] = {"maximum": 0.99} + + # Asimov's RIFT PSD convention is sample-rate -> detector -> path, while + # Rimsky records detector -> path. This copy belongs only to the RIFT + # production, so the event document retained by Rimsky remains unchanged. + psds = normalized.get("psds") + sample_rate = normalized.get("likelihood", {}).get("sample rate") + detectors = normalized.get("interferometers", []) + if ( + isinstance(psds, dict) + and sample_rate is not None + and detectors + and all(detector in psds for detector in detectors) + ): + normalized["psds"] = { + sample_rate: { + detector: copy.deepcopy(psds[detector]) for detector in detectors + } + } + return normalized + + +def write_analysis(analysis, path): + """Write one analysis document as YAML (or JSON, which is valid YAML).""" + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + try: + import yaml + except ImportError: + with path.open("w", encoding="utf-8") as stream: + json.dump(analysis, stream, indent=2) + stream.write("\n") + else: + with path.open("w", encoding="utf-8") as stream: + yaml.safe_dump(analysis, stream, sort_keys=False) + return path + + +def main(argv=None): + """Command-line entry point for creating a Rimsky RIFT follow-up file.""" + parser = argparse.ArgumentParser( + description="Generate a RIFT follow-up analysis for a Rimsky configuration" + ) + parser.add_argument("rimsky_config", help="Rimsky YAML configuration") + parser.add_argument("output", help="Destination analysis YAML") + args = parser.parse_args(argv) + + config = load_rimsky_config(args.rimsky_config) + path = write_analysis( + build_analysis(config, config_path=args.rimsky_config), args.output + ) + print(path) + + +if __name__ == "__main__": + main() diff --git a/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py b/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py new file mode 100644 index 000000000..bec996cd6 --- /dev/null +++ b/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py @@ -0,0 +1,248 @@ +import copy +import configparser +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from RIFT.rimsky import ( + RimskyIntegrationError, + build_analysis, + normalize_event_metadata, + write_analysis, +) + + +def _rimsky_config(tmp_path): + return { + "detectors": ["H1", "L1", "V1"], + "output_dir": "online-output", + "event_sink": { + "bilby_pipe_defaults": { + "minimum_frequency": 18, + "maximum_frequency": {"H1": 1024, "L1": 1024, "V1": 896}, + } + }, + "rift": { + "name": "rift-low-latency", + "waveform": {"approximant": "SEOBNRv5PHM"}, + }, + } + + +def test_build_analysis_targets_rimsky_pesummary_output(tmp_path): + config_path = tmp_path / "configs" / "rimsky.yaml" + analysis = build_analysis(_rimsky_config(tmp_path), config_path=config_path) + + expected = ( + config_path.parent + / "online-output" + / "*" + / "*" + / "{event}" + / "results_page" + / "metafile.hdf5" + ).resolve() + assert analysis["kind"] == "analysis" + assert analysis["pipeline"] == "RIFT" + assert analysis["name"] == "rift-low-latency" + assert analysis["dataset"] == "bilby-online" + assert analysis["scheduler"]["bootstrap file"] == str(expected) + assert analysis["quality"]["minimum frequency"] == { + "H1": 18, + "L1": 18, + "V1": 18, + } + assert analysis["quality"]["maximum frequency"]["V1"] == 896 + assert analysis["waveform"]["approximant"] == "SEOBNRv5PHM" + + +def test_normalize_rimsky_event_priors_is_additive(): + event = { + "name": "S260305df", + "interferometers": ["H1", "L1"], + "likelihood": {"sample rate": 2048}, + "psds": {"H1": "/tmp/H1.txt", "L1": "/tmp/L1.txt"}, + "priors": { + "chirp_mass": {"minimum": 10, "maximum": 20}, + "mass_ratio": {"minimum": 0.1, "maximum": 1}, + "luminosity_distance": { + "minimum": 10, + "maximum": 5000, + "type": "bilby.gw.prior.UniformSourceFrame", + }, + "a_1": {"minimum": 0, "maximum": 0.8}, + "a_2": {"minimum": 0, "maximum": 0.7}, + }, + } + original = copy.deepcopy(event) + normalized = normalize_event_metadata(event) + + assert event == original + assert normalized["priors"]["chirp mass"] == event["priors"]["chirp_mass"] + assert normalized["priors"]["mass ratio"] == event["priors"]["mass_ratio"] + assert normalized["priors"]["luminosity distance"]["maximum"] == 5000 + assert normalized["priors"]["spin 1"] == {"maximum": 0.8} + assert normalized["priors"]["spin 2"] == {"maximum": 0.7} + assert "a_1" in normalized["priors"] + assert normalized["psds"] == {2048: {"H1": "/tmp/H1.txt", "L1": "/tmp/L1.txt"}} + + +def test_normalize_does_not_replace_explicit_rift_prior(): + event = {"priors": {"chirp_mass": {"maximum": 20}, "chirp mass": {"maximum": 30}}} + assert normalize_event_metadata(event)["priors"]["chirp mass"]["maximum"] == 30 + + +def test_rift_pipeline_normalizes_rimsky_metadata_before_templating(): + from RIFT.asimov.rift import Rift + + pipeline = object.__new__(Rift) + pipeline.production = SimpleNamespace( + meta={ + "priors": { + "chirp_mass": {"minimum": 10, "maximum": 20}, + "a_1": {"maximum": 0.8}, + }, + "likelihood": {}, + } + ) + pipeline._create_ledger_entries() + + assert pipeline.production.meta["priors"]["chirp mass"]["maximum"] == 20 + assert pipeline.production.meta["priors"]["spin 1"]["maximum"] == 0.8 + assert pipeline.production.meta["sampler"] == {"cip": {}, "ile": {}} + assert pipeline.production.meta["likelihood"] == { + "assume": {}, + "marginalization": {}, + } + + +def test_rift_pipeline_builds_lal_caches_for_rimsky_frames(tmp_path): + from RIFT.asimov.rift import Rift + + frames = [] + for start in (1456739148, 1456739152): + frame = tmp_path / "S260305df-H1-{}-4.gwf".format(start) + frame.touch() + frames.append(str(frame)) + + work_dir = tmp_path / "work" + pipeline = object.__new__(Rift) + pipeline.production = SimpleNamespace( + name="rift-online", + event=SimpleNamespace(work_dir=str(work_dir)), + meta={"data": {"data files": {"H1": frames}}}, + ) + caches = pipeline._prepare_frame_caches() + + cache = Path(caches["H1"]) + assert cache == work_dir / "H1-rimsky.cache" + lines = cache.read_text().splitlines() + assert lines == [ + "H RIMSKY 1456739148 4 {}".format(Path(frames[0]).as_uri()), + "H RIMSKY 1456739152 4 {}".format(Path(frames[1]).as_uri()), + ] + assert pipeline.production.meta["data"]["frame cache"] == caches + + +def test_rift_template_passes_generated_frame_caches(): + template = ( + Path(__file__).resolve().parents[1] / "RIFT" / "asimov" / "rift.ini" + ).read_text() + assert "fake-cache" in template + assert "data['frame cache'][ifo]" in template + + +def test_generated_rimsky_analysis_renders_rift_template(tmp_path): + liquid = pytest.importorskip("liquid") + analysis = build_analysis( + _rimsky_config(tmp_path), config_path=tmp_path / "rimsky.yaml" + ) + event = { + "engine": "RIFT", + "interferometers": ["H1", "L1", "V1"], + "data": { + "segment length": 8, + "channels": {ifo: "{}:STRAIN".format(ifo) for ifo in ("H1", "L1", "V1")}, + "frame types": {ifo: "gwf" for ifo in ("H1", "L1", "V1")}, + "frame cache": { + ifo: "/tmp/{}-rimsky.cache".format(ifo) for ifo in ("H1", "L1", "V1") + }, + }, + "likelihood": {"sample rate": 2048}, + "priors": { + "chirp_mass": {"minimum": 10, "maximum": 20}, + "mass_ratio": {"minimum": 0.1, "maximum": 1}, + "luminosity_distance": { + "minimum": 10, + "maximum": 5000, + "type": "bilby.gw.prior.UniformSourceFrame", + }, + "a_1": {"maximum": 0.8}, + "a_2": {"maximum": 0.7}, + }, + } + for key, value in analysis.items(): + if isinstance(value, dict) and isinstance(event.get(key), dict): + event[key].update(copy.deepcopy(value)) + else: + event[key] = copy.deepcopy(value) + meta = normalize_event_metadata(event) + + production = SimpleNamespace( + name=meta["name"], + meta=meta, + category="C01_offline", + event=SimpleNamespace(name="S260305df", repository=None), + xml_psds={ + ifo: "/tmp/{}-psd.xml.gz".format(ifo) for ifo in meta["interferometers"] + }, + ) + context = { + "production": production, + "config": { + "general": {"webroot": "/tmp/rift-web"}, + "pipelines": {"environment": "/opt/igwn"}, + "condor": {"user": "riftci"}, + }, + } + template_text = ( + Path(__file__).resolve().parents[1] / "RIFT" / "asimov" / "rift.ini" + ).read_text() + if hasattr(liquid, "Environment"): + rendered = liquid.Environment().from_string(template_text).render(**context) + elif hasattr(liquid, "Liquid"): + rendered = liquid.Liquid(template_text, from_file=False).render(**context) + else: + rendered = liquid.Template(template_text).render(**context) + + parser = configparser.RawConfigParser() + parser.read_string(rendered) + assert parser.get("engine", "chirpmass-min") == "10" + assert parser.get("engine", "comp-max") == "1000" + assert parser.get("engine", "a_spin1-max") == "0.8" + assert '"V1":"/tmp/V1-rimsky.cache"' in parser.get("lalinference", "fake-cache") + + +def test_write_analysis_round_trips(tmp_path): + yaml = pytest.importorskip("yaml") + analysis = build_analysis( + _rimsky_config(tmp_path), config_path=tmp_path / "rimsky.yaml" + ) + destination = write_analysis(analysis, tmp_path / "rift-followup.yaml") + assert yaml.safe_load(destination.read_text()) == analysis + + +@pytest.mark.parametrize("detectors", [[], {"H1": "bad"}, ["H1", 2]]) +def test_invalid_detectors_fail_early(tmp_path, detectors): + config = _rimsky_config(tmp_path) + config["detectors"] = detectors + with pytest.raises(RimskyIntegrationError, match="detector"): + build_analysis(config, config_path=tmp_path / "rimsky.yaml") + + +def test_frequency_mapping_must_cover_all_detectors(tmp_path): + config = _rimsky_config(tmp_path) + config["event_sink"]["bilby_pipe_defaults"]["maximum_frequency"] = {"H1": 1024} + with pytest.raises(RimskyIntegrationError, match="L1, V1"): + build_analysis(config, config_path=tmp_path / "rimsky.yaml") diff --git a/setup.py b/setup.py index bfd0b326c..5b816b083 100644 --- a/setup.py +++ b/setup.py @@ -59,6 +59,8 @@ setup_requires=['setuptools','pip'], install_requires=REQUIREMENTS["install"], entry_points={ + 'console_scripts': + ['rift-rimsky-analysis = RIFT.rimsky.integration:main'], 'asimov.pipelines': ["rift = RIFT.asimov.rift:Rift"], 'RIFT.integrator_plugins': From 222bcccd602376f0f872ae9d7e3e002be529e619 Mon Sep 17 00:00:00 2001 From: Richard O'Shaughnessy Date: Wed, 2 Sep 2026 09:17:15 -0700 Subject: [PATCH 2/9] Scope frame caches to Rimsky analyses --- .../Code/RIFT/asimov/rift.py | 2 ++ .../Code/RIFT/rimsky/integration.py | 1 + .../Code/test/test_rimsky_integration.py | 19 ++++++++++++++++++- 3 files changed, 21 insertions(+), 1 deletion(-) diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py index ea11ae66f..4d423b7d3 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py +++ b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py @@ -100,6 +100,8 @@ def _get_psds(self, format="ascii"): 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: diff --git a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py index 37c699049..5ec583f50 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py +++ b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py @@ -117,6 +117,7 @@ def build_analysis(config, *, config_path=None, overrides=None): "name": "rift-online", "status": "Ready", "pipeline": "RIFT", + "orchestrator": "rimsky", "comment": "RIFT follow-up launched by Rimsky after online Bilby PE", "dataset": "bilby-online", "likelihood": { diff --git a/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py b/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py index bec996cd6..c4d7b7b6b 100644 --- a/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py +++ b/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py @@ -45,6 +45,7 @@ def test_build_analysis_targets_rimsky_pesummary_output(tmp_path): ).resolve() assert analysis["kind"] == "analysis" assert analysis["pipeline"] == "RIFT" + assert analysis["orchestrator"] == "rimsky" assert analysis["name"] == "rift-low-latency" assert analysis["dataset"] == "bilby-online" assert analysis["scheduler"]["bootstrap file"] == str(expected) @@ -131,7 +132,10 @@ def test_rift_pipeline_builds_lal_caches_for_rimsky_frames(tmp_path): pipeline.production = SimpleNamespace( name="rift-online", event=SimpleNamespace(work_dir=str(work_dir)), - meta={"data": {"data files": {"H1": frames}}}, + meta={ + "orchestrator": "rimsky", + "data": {"data files": {"H1": frames}}, + }, ) caches = pipeline._prepare_frame_caches() @@ -145,6 +149,19 @@ def test_rift_pipeline_builds_lal_caches_for_rimsky_frames(tmp_path): assert pipeline.production.meta["data"]["frame cache"] == caches +def test_frame_cache_generation_is_isolated_to_rimsky(tmp_path): + from RIFT.asimov.rift import Rift + + pipeline = object.__new__(Rift) + pipeline.production = SimpleNamespace( + name="unrelated-analysis", + event=SimpleNamespace(work_dir=str(tmp_path)), + meta={"data": {"data files": {"H1": ["not-a-rimsky-frame"]}}}, + ) + assert pipeline._prepare_frame_caches() == {} + assert "frame cache" not in pipeline.production.meta["data"] + + def test_rift_template_passes_generated_frame_caches(): template = ( Path(__file__).resolve().parents[1] / "RIFT" / "asimov" / "rift.ini" From ca70f1e0ffe40f4a08e92314738da0949ff402e1 Mon Sep 17 00:00:00 2001 From: Richard O'Shaughnessy Date: Wed, 2 Sep 2026 09:28:52 -0700 Subject: [PATCH 3/9] Add automatic Rimsky follow-up and E2E test --- .../Code/RIFT/rimsky/README.md | 29 ++- .../Code/RIFT/rimsky/__init__.py | 4 + .../Code/RIFT/rimsky/integration.py | 73 +++++++- .../Code/test/test_rimsky_end_to_end.py | 172 ++++++++++++++++++ .../Code/test/test_rimsky_integration.py | 33 ++++ 5 files changed, 299 insertions(+), 12 deletions(-) create mode 100644 MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py diff --git a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md index 0f1f4f697..8f55b0c9f 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md +++ b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md @@ -4,11 +4,18 @@ 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 a RIFT Asimov analysis document. -2. Set `sample_sink.asimov_configuration` in `rimsky.yaml` to the absolute path - of `rift-followup.yaml`. -3. Set `asimovdir` to an initialized Asimov project in which the RIFT package is - installed and its pipeline is configured. + configuration and writes both a RIFT Asimov analysis document and a runnable + `rimsky-rift.yaml`. +2. Initialize the Asimov 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: @@ -18,6 +25,7 @@ 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. @@ -41,7 +49,10 @@ Rimsky 0.1 event documents use Bilby-style prior names (`chirp_mass`, the space-separated aliases expected by its Asimov template. This makes the same event usable by both Bilby and RIFT analyses. -The bridge consumes plain YAML mappings and does not import Rimsky. It is -therefore lightweight to test and isolated from Rimsky's streaming, GraceDB, -and HTCondor dependencies. The contract targets Rimsky `0.1.0rc1` and current -main as of 2026-09-02. +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. 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). diff --git a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py index c75962a30..6020ad1de 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py +++ b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py @@ -9,15 +9,19 @@ from .integration import ( RimskyIntegrationError, build_analysis, + configure_rimsky, load_rimsky_config, normalize_event_metadata, write_analysis, + write_rimsky_config, ) __all__ = [ "RimskyIntegrationError", "build_analysis", + "configure_rimsky", "load_rimsky_config", "normalize_event_metadata", "write_analysis", + "write_rimsky_config", ] diff --git a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py index 5ec583f50..c45a41765 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py +++ b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py @@ -74,6 +74,15 @@ def _resolve_output_dir(config, config_path=None): return output_dir.resolve() +def _resolve_config_path(value, *, config_path=None): + """Resolve a Rimsky path using the source configuration as its anchor.""" + path = Path(value).expanduser() + if not path.is_absolute(): + base = Path(config_path).resolve().parent if config_path else Path.cwd() + path = base / path + return path.resolve() + + def build_analysis(config, *, config_path=None, overrides=None): """Build the Asimov analysis document consumed by Rimsky's follow-up hook. @@ -160,6 +169,41 @@ def build_analysis(config, *, config_path=None, overrides=None): return analysis +def configure_rimsky(config, analysis_path, *, config_path=None): + """Return a runnable Rimsky configuration wired to the RIFT follow-up. + + Rimsky interprets paths relative to its launch directory, rather than the + YAML file. Emit absolute paths so the online output observed by Rimsky is + the same output searched by RIFT's bootstrap glob. Missing orchestration + settings default to a local Asimov project and a submitted Bilby run; an + operator's explicit values are retained. + """ + if not isinstance(config, dict): + raise RimskyIntegrationError("Rimsky configuration must be a mapping") + + configured = copy.deepcopy(config) + configured["output_dir"] = str( + _resolve_output_dir(configured, config_path=config_path) + ) + configured["asimovdir"] = str( + _resolve_config_path( + configured.get("asimovdir", "asimov"), config_path=config_path + ) + ) + event_sink = configured.setdefault("event_sink", {}) + if not isinstance(event_sink, dict): + raise RimskyIntegrationError("Rimsky 'event_sink' must be a mapping") + event_sink.setdefault("bilby_pipe_format", "full-submit") + + sample_sink = configured.setdefault("sample_sink", {}) + if not isinstance(sample_sink, dict): + raise RimskyIntegrationError("Rimsky 'sample_sink' must be a mapping") + sample_sink["asimov_configuration"] = str( + _resolve_config_path(analysis_path) + ) + return configured + + def normalize_event_metadata(metadata): """Return RIFT-compatible metadata from a Rimsky-created event mapping. @@ -224,20 +268,43 @@ def write_analysis(analysis, path): return path +def write_rimsky_config(config, path): + """Write a Rimsky configuration containing the automatic RIFT hook.""" + return write_analysis(config, path) + + def main(argv=None): - """Command-line entry point for creating a Rimsky RIFT follow-up file.""" + """Create the RIFT analysis and a Rimsky config that invokes it.""" parser = argparse.ArgumentParser( description="Generate a RIFT follow-up analysis for a Rimsky configuration" ) parser.add_argument("rimsky_config", help="Rimsky YAML configuration") parser.add_argument("output", help="Destination analysis YAML") + parser.add_argument( + "--configured-rimsky", + help=( + "Destination for the runnable Rimsky YAML " + "(default: -rift.yaml)" + ), + ) args = parser.parse_args(argv) config = load_rimsky_config(args.rimsky_config) - path = write_analysis( + analysis_path = write_analysis( build_analysis(config, config_path=args.rimsky_config), args.output ) - print(path) + source = Path(args.rimsky_config) + configured_path = Path(args.configured_rimsky) if args.configured_rimsky else ( + source.with_name(source.stem + "-rift" + source.suffix) + ) + write_rimsky_config( + configure_rimsky( + config, analysis_path.resolve(), config_path=args.rimsky_config + ), + configured_path, + ) + print(analysis_path) + print(configured_path) if __name__ == "__main__": diff --git a/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py b/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py new file mode 100644 index 000000000..2345265a8 --- /dev/null +++ b/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py @@ -0,0 +1,172 @@ +"""End-to-end contract test for Rimsky's real Asimov follow-up hook. + +This test deliberately stops before submitting external HTCondor jobs. It +does exercise both installed projects, the YAML files exchanged between them, +Asimov's ledger, RIFT pipeline discovery, and bootstrap-file resolution. +""" + +import configparser +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import yaml + +rimsky = pytest.importorskip("rimsky", reason="Rimsky requires Python >=3.12") + +import asimov +from asimov.ledger import YAMLLedger +from asimov.utils import update +from rimsky.settings import PipelineSettings +from rimsky.sinks.gdb_samples import start_asimov +from rimsky.utils.asimov import add_event + +from RIFT.rimsky.integration import main + + +ASIMOV_CONFIG = """ +[ledger] +location = ledger.yaml +engine = yamlfile + +[project] +name = rimsky-rift-e2e +root = {project} + +[logging] +level = info +directory = logs +location = logs/asimov.log + +[pipelines] +environment = test + +[general] +git_default = . +rundir_default = {project}/working +calibration = test +calibration_directory = test +webroot = pages/ +logger = file +""" + + +def _initialise_asimov(project): + project.mkdir() + ledger_path = project / "ledger.yaml" + config = configparser.ConfigParser() + config.read_string(ASIMOV_CONFIG.format(project=project)) + asimov.config = config + asimov.analysis.config = config + asimov.event.config = config + asimov.ledger.config = config + YAMLLedger.create(location=ledger_path, name="rimsky-rift-e2e") + ledger = YAMLLedger(location=str(ledger_path)) + update(ledger.data, {"pipelines": {"rift": {}}}) + asimov.current_ledger = ledger + return ledger + + +def test_first_rimsky_result_creates_bootstrapped_rift_production(tmp_path): + sid = "S260305df" + source = tmp_path / "rimsky.yaml" + followup = tmp_path / "rift-followup.yaml" + configured_path = tmp_path / "rimsky-rift.yaml" + source.write_text( + yaml.safe_dump( + { + "detectors": ["H1", "L1"], + "channels": {"H1": "STRAIN", "L1": "STRAIN"}, + "output_dir": "online-output", + "event_sink": { + "bilby_pipe_defaults": { + "minimum_frequency": 20, + "maximum_frequency": 1024, + }, + "trigger_dependent_settings": {}, + "prior_defaults": {}, + }, + }, + sort_keys=False, + ), + encoding="utf-8", + ) + + main( + [ + str(source), + str(followup), + "--configured-rimsky", + str(configured_path), + ] + ) + + # Load the generated file through Rimsky itself. These are the defaults + # which make Bilby run first and the sample sink enqueue RIFT afterwards. + settings = PipelineSettings.from_yaml(configured_path) + assert settings.event_sink.bilby_pipe_format == "full-submit" + assert Path(settings.sample_sink.asimov_configuration) == followup + assert Path(settings.asimovdir) == tmp_path / "asimov" + + result = ( + Path(settings.output_dir) + / sid[1:5] + / sid[5:7] + / sid + / "results_page" + / "metafile.hdf5" + ) + result.parent.mkdir(parents=True) + result.touch() + + frames = {} + psds = {} + for detector in settings.detectors: + frame = tmp_path / "{}-RIMSKY-1456739148-4.gwf".format(detector) + frame.touch() + frames[detector] = [str(frame)] + psd = tmp_path / "{}-psd.txt".format(detector) + psd.touch() + psds[detector] = str(psd) + + ledger = _initialise_asimov(Path(settings.asimovdir)) + event_metadata = { + "name": sid, + "category": "online", + "interferometers": settings.detectors, + "data": { + "segment length": 8, + "channels": settings.channels, + "data files": frames, + }, + "likelihood": {"sample rate": 2048}, + "psds": psds, + "priors": { + "chirp_mass": {"minimum": 10, "maximum": 20}, + "mass_ratio": {"minimum": 0.1, "maximum": 1}, + "luminosity_distance": {"minimum": 10, "maximum": 5000}, + "a_1": {"minimum": 0, "maximum": 0.8}, + "a_2": {"minimum": 0, "maximum": 0.8}, + }, + } + with patch("git.Repo", return_value=MagicMock()): + add_event(Path(settings.asimovdir), event_metadata, ledger=ledger) + start_asimov( + event=sid, + asimovdir=Path(settings.asimovdir), + asimov_configuration=settings.sample_sink.asimov_configuration, + ) + event = ledger.get_event(sid)[0] + productions = [ + item for item in event.analyses if item.name == "rift-online" + ] + assert len(productions) == 1 + production = productions[0] + pipeline = production.pipeline + assert pipeline.__class__.__name__ == "Rift" + assert pipeline._resolve_bootstrap_file() == str(result) + assert production.meta["priors"]["chirp mass"]["maximum"] == 20 + assert production.meta["psds"] == {2048: psds} + caches = pipeline._prepare_frame_caches() + assert set(caches) == {"H1", "L1"} + assert all(Path(cache).is_file() for cache in caches.values()) diff --git a/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py b/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py index c4d7b7b6b..597f65267 100644 --- a/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py +++ b/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py @@ -5,9 +5,14 @@ import pytest +# Import Asimov before its RIFT entry point. Importing the entry-point module +# first creates a circular discovery path in Asimov 0.5/0.6. +pytest.importorskip("asimov") + from RIFT.rimsky import ( RimskyIntegrationError, build_analysis, + configure_rimsky, normalize_event_metadata, write_analysis, ) @@ -250,6 +255,34 @@ def test_write_analysis_round_trips(tmp_path): assert yaml.safe_load(destination.read_text()) == analysis +def test_configure_rimsky_defaults_to_submitted_bilby_then_rift(tmp_path): + source = tmp_path / "configs" / "rimsky.yaml" + followup = tmp_path / "generated" / "rift-followup.yaml" + configured = configure_rimsky( + _rimsky_config(tmp_path), followup, config_path=source + ) + + assert configured["output_dir"] == str( + (source.parent / "online-output").resolve() + ) + assert configured["asimovdir"] == str((source.parent / "asimov").resolve()) + assert configured["event_sink"]["bilby_pipe_format"] == "full-submit" + assert configured["sample_sink"]["asimov_configuration"] == str( + followup.resolve() + ) + + +def test_configure_rimsky_preserves_explicit_run_mode_and_asimovdir(tmp_path): + config = _rimsky_config(tmp_path) + config["asimovdir"] = "project" + config["event_sink"]["bilby_pipe_format"] = "full-local" + configured = configure_rimsky( + config, tmp_path / "followup.yaml", config_path=tmp_path / "rimsky.yaml" + ) + assert configured["event_sink"]["bilby_pipe_format"] == "full-local" + assert configured["asimovdir"] == str((tmp_path / "project").resolve()) + + @pytest.mark.parametrize("detectors", [[], {"H1": "bad"}, ["H1", 2]]) def test_invalid_detectors_fail_early(tmp_path, detectors): config = _rimsky_config(tmp_path) From cbbcd7a4c5419b6965095b6f4de4093d924099c1 Mon Sep 17 00:00:00 2001 From: Richard O'Shaughnessy Date: Wed, 2 Sep 2026 09:29:59 -0700 Subject: [PATCH 4/9] Run Rimsky integration against release and main --- .github/workflows/rimsky-integration.yml | 43 ++++++++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 .github/workflows/rimsky-integration.yml diff --git a/.github/workflows/rimsky-integration.yml b/.github/workflows/rimsky-integration.yml new file mode 100644 index 000000000..06bcfbb9e --- /dev/null +++ b/.github/workflows/rimsky-integration.yml @@ -0,0 +1,43 @@ +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: | + python -m pip install --upgrade pip --break-system-packages + python -m pip install -r requirements.txt --break-system-packages + python -m pip install pytest --break-system-packages + python -m pip install --editable . --break-system-packages + python -m pip install 'asimov==0.6.1' 'htcondor<25' '${{ matrix.rimsky-spec }}' --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 From f434e1fa895f07fff637fc4c29319a383cefd05a Mon Sep 17 00:00:00 2001 From: Richard O'Shaughnessy Date: Wed, 2 Sep 2026 09:39:15 -0700 Subject: [PATCH 5/9] Require Asimov 0.7 in Rimsky E2E --- MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md | 12 +++++++----- .../Code/test/test_rimsky_end_to_end.py | 4 ++++ 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md index 8f55b0c9f..b29a97f64 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md +++ b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md @@ -6,7 +6,7 @@ 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 the Asimov project named by `asimovdir` once, install RIFT in its +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 @@ -52,7 +52,9 @@ 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. 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). +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. diff --git a/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py b/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py index 2345265a8..8b83116c7 100644 --- a/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py +++ b/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py @@ -6,11 +6,13 @@ """ import configparser +from importlib.metadata import version from pathlib import Path from unittest.mock import MagicMock, patch import pytest import yaml +from packaging.version import Version rimsky = pytest.importorskip("rimsky", reason="Rimsky requires Python >=3.12") @@ -68,6 +70,8 @@ def _initialise_asimov(project): def test_first_rimsky_result_creates_bootstrapped_rift_production(tmp_path): + assert Version(version("asimov")) >= Version("0.7") + sid = "S260305df" source = tmp_path / "rimsky.yaml" followup = tmp_path / "rift-followup.yaml" From 22979352eafc2e22e8e7006072f15fe6b8fa5b4c Mon Sep 17 00:00:00 2001 From: Richard O'Shaughnessy Date: Wed, 2 Sep 2026 09:40:29 -0700 Subject: [PATCH 6/9] Test and support Rimsky on Asimov 0.7 --- .github/workflows/rimsky-integration.yml | 3 ++- .../Code/RIFT/asimov/rift.py | 20 ++++++++++++++++--- 2 files changed, 19 insertions(+), 4 deletions(-) diff --git a/.github/workflows/rimsky-integration.yml b/.github/workflows/rimsky-integration.yml index 06bcfbb9e..8d1657f00 100644 --- a/.github/workflows/rimsky-integration.yml +++ b/.github/workflows/rimsky-integration.yml @@ -38,6 +38,7 @@ jobs: python -m pip install -r requirements.txt --break-system-packages python -m pip install pytest --break-system-packages python -m pip install --editable . --break-system-packages - python -m pip install 'asimov==0.6.1' 'htcondor<25' '${{ matrix.rimsky-spec }}' --break-system-packages + python -m pip install '${{ matrix.rimsky-spec }}' --break-system-packages + python -m pip install --upgrade 'asimov>=0.7,<0.8' 'bilby_pipe @ git+https://git.ligo.org/lscsoft/bilby_pipe.git@be6c77021db809690c781a7744f40564e62dd72f' --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 diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py index 4d423b7d3..44529747a 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py +++ b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py @@ -13,7 +13,12 @@ from asimov.utils import set_directory from asimov.pipeline import Pipeline, PipelineException, PipelineLogger -from asimov.pipeline import PESummaryPipeline + +try: + from asimov.pipeline import PESummaryPipeline +except ImportError: + # Asimov >= 0.7 supplies PESummary as a separate pipeline plugin. + PESummaryPipeline = None from asimov.utils import update @@ -277,9 +282,18 @@ def _find_posterior(self): self.logger.error("Could not find an analysis providing posterior samples to analyse.") def after_completion(self): + if PESummaryPipeline is None: + self.logger.info( + "Job has completed. PESummary is managed by a separate " + "Asimov postprocessing analysis." + ) + super().after_completion() + return - self.logger.info("Job has completed. Running PE Summary.") - post_pipeline = PESummaryPipeline(production=self.production) + self.logger.info("Job has completed. Running legacy PE Summary.") + post_pipeline = PESummaryPipeline( + production=self.production, category=self.category + ) cluster = post_pipeline.submit_dag() self.production.meta["job id"] = int(cluster) From a92e02643d329d5d18a3582d55207f042a7fc331 Mon Sep 17 00:00:00 2001 From: Richard O'Shaughnessy Date: Wed, 2 Sep 2026 09:42:45 -0700 Subject: [PATCH 7/9] Install only E2E orchestration dependencies --- .github/workflows/rimsky-integration.yml | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/.github/workflows/rimsky-integration.yml b/.github/workflows/rimsky-integration.yml index 8d1657f00..ebae8da76 100644 --- a/.github/workflows/rimsky-integration.yml +++ b/.github/workflows/rimsky-integration.yml @@ -35,10 +35,9 @@ jobs: - name: Install RIFT and Rimsky orchestration stack run: | python -m pip install --upgrade pip --break-system-packages - python -m pip install -r requirements.txt --break-system-packages - python -m pip install pytest --break-system-packages - python -m pip install --editable . --break-system-packages python -m pip install '${{ matrix.rimsky-spec }}' --break-system-packages python -m pip install --upgrade 'asimov>=0.7,<0.8' 'bilby_pipe @ git+https://git.ligo.org/lscsoft/bilby_pipe.git@be6c77021db809690c781a7744f40564e62dd72f' --break-system-packages + 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 From 73126418870118bb9c1cb363fcbb13fbca3fbebf Mon Sep 17 00:00:00 2001 From: Richard O'Shaughnessy Date: Wed, 2 Sep 2026 09:47:31 -0700 Subject: [PATCH 8/9] Retry transient Rimsky dependency downloads --- .github/workflows/rimsky-integration.yml | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/.github/workflows/rimsky-integration.yml b/.github/workflows/rimsky-integration.yml index ebae8da76..61c162cc6 100644 --- a/.github/workflows/rimsky-integration.yml +++ b/.github/workflows/rimsky-integration.yml @@ -34,9 +34,20 @@ jobs: 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 - python -m pip install '${{ matrix.rimsky-spec }}' --break-system-packages - python -m pip install --upgrade 'asimov>=0.7,<0.8' 'bilby_pipe @ git+https://git.ligo.org/lscsoft/bilby_pipe.git@be6c77021db809690c781a7744f40564e62dd72f' --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 From 9221e6457bd77e0c9e568066b51c5d77ee847bd7 Mon Sep 17 00:00:00 2001 From: Richard O'Shaughnessy Date: Fri, 4 Sep 2026 04:34:08 -0700 Subject: [PATCH 9/9] Exercise Rimsky through Asimov submission --- .../Code/RIFT/asimov/rift.ini | 9 +- .../Code/RIFT/asimov/rift.py | 106 ++++++++++++-- .../Code/test/test_rimsky_end_to_end.py | 137 ++++++++++++++++-- 3 files changed, 222 insertions(+), 30 deletions(-) diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini index 30b6ccf06..ccc3a225a 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini +++ b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini @@ -62,7 +62,9 @@ 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 %} } @@ -296,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" diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py index b3c77ec61..be2ff3f2f 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py +++ b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py @@ -103,6 +103,73 @@ def _get_psds(self, format="ascii"): 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": @@ -357,30 +424,37 @@ def before_config(self, dryrun=False): 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}.", ) - xml_psds = getattr(self.production, "xml_psds", None) - if isinstance(xml_psds, dict): - xml_psds[ifo] = os.path.join( - self.production.event.repository.directory, saveloc - ) - 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']: @@ -732,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") @@ -775,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 = [ diff --git a/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py b/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py index 8b83116c7..897b2e28f 100644 --- a/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py +++ b/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py @@ -1,15 +1,21 @@ -"""End-to-end contract test for Rimsky's real Asimov follow-up hook. +"""Submission-level test for Rimsky's real Asimov follow-up hook. -This test deliberately stops before submitting external HTCondor jobs. It -does exercise both installed projects, the YAML files exchanged between them, -Asimov's ledger, RIFT pipeline discovery, and bootstrap-file resolution. +The test exercises both installed projects, their exchanged YAML, Asimov's +ledger and configuration rendering, and RIFT's real input discovery and +conversion. It replaces only the heavyweight pseudo-pipeline process and the +HTCondor scheduler boundary. """ import configparser +import os +import sys +from contextlib import chdir from importlib.metadata import version from pathlib import Path from unittest.mock import MagicMock, patch +import h5py +import numpy as np import pytest import yaml from packaging.version import Version @@ -21,8 +27,9 @@ from asimov.utils import update from rimsky.settings import PipelineSettings from rimsky.sinks.gdb_samples import start_asimov -from rimsky.utils.asimov import add_event +from rimsky.utils.asimov import add_event, build_and_submit +import RIFT.asimov.rift as rift_asimov from RIFT.rimsky.integration import main @@ -41,13 +48,16 @@ location = logs/asimov.log [pipelines] -environment = test +environment = {environment} + +[condor] +user = rimsky-test [general] git_default = . rundir_default = {project}/working calibration = test -calibration_directory = test +calibration_directory = C01_offline webroot = pages/ logger = file """ @@ -57,11 +67,14 @@ def _initialise_asimov(project): project.mkdir() ledger_path = project / "ledger.yaml" config = configparser.ConfigParser() - config.read_string(ASIMOV_CONFIG.format(project=project)) + config.read_string( + ASIMOV_CONFIG.format(project=project, environment=sys.prefix) + ) asimov.config = config asimov.analysis.config = config asimov.event.config = config asimov.ledger.config = config + rift_asimov.config = config YAMLLedger.create(location=ledger_path, name="rimsky-rift-e2e") ledger = YAMLLedger(location=str(ledger_path)) update(ledger.data, {"pipelines": {"rift": {}}}) @@ -69,7 +82,9 @@ def _initialise_asimov(project): return ledger -def test_first_rimsky_result_creates_bootstrapped_rift_production(tmp_path): +def test_first_rimsky_result_creates_bootstrapped_rift_production( + tmp_path, monkeypatch +): assert Version(version("asimov")) >= Version("0.7") sid = "S260305df" @@ -121,7 +136,33 @@ def test_first_rimsky_result_creates_bootstrapped_rift_production(tmp_path): / "metafile.hdf5" ) result.parent.mkdir(parents=True) - result.touch() + posterior = np.zeros( + 2, + dtype=[ + ("mass_1", "f8"), + ("mass_2", "f8"), + ("chirp_mass", "f8"), + ("luminosity_distance", "f8"), + ("phase", "f8"), + ("iota", "f8"), + ("spin_1x", "f8"), + ("spin_1y", "f8"), + ("spin_1z", "f8"), + ("spin_2x", "f8"), + ("spin_2y", "f8"), + ("spin_2z", "f8"), + ], + ) + posterior["mass_1"] = [35, 36] + posterior["mass_2"] = [30, 29] + posterior["chirp_mass"] = [28, 27] + posterior["luminosity_distance"] = [400, 420] + posterior["iota"] = [0.5, 0.6] + posterior["spin_1z"] = [0.1, 0.2] + posterior["spin_2z"] = [-0.1, -0.2] + with h5py.File(result, "w") as metafile: + analysis = metafile.create_group("bilby-online") + analysis.create_dataset("posterior_samples", data=posterior) frames = {} psds = {} @@ -130,7 +171,7 @@ def test_first_rimsky_result_creates_bootstrapped_rift_production(tmp_path): frame.touch() frames[detector] = [str(frame)] psd = tmp_path / "{}-psd.txt".format(detector) - psd.touch() + np.savetxt(psd, [[0, 1e-40], [1, 1e-40], [2, 1e-40]]) psds[detector] = str(psd) ledger = _initialise_asimov(Path(settings.asimovdir)) @@ -174,3 +215,77 @@ def test_first_rimsky_result_creates_bootstrapped_rift_production(tmp_path): caches = pipeline._prepare_frame_caches() assert set(caches) == {"H1", "L1"} assert all(Path(cache).is_file() for cache in caches.values()) + + # Exercise the same input-discovery and template-rendering hook used by + # ``asimov manage build``. Keep the installed RIFT scripts discoverable + # when this test is launched via an explicit virtual-environment Python. + monkeypatch.setenv( + "PATH", "{}:{}".format(Path(sys.executable).parent, os.environ["PATH"]) + ) + project_dir = Path(settings.asimovdir) + with chdir(project_dir), patch("asimov.git.time.sleep"): + pipeline.before_config() + + for detector in settings.detectors: + xml_psd = Path(production.xml_psds[detector]) + assert Path(xml_psd).is_file() + assert set(production.meta["data"]["frame cache"]) == {"H1", "L1"} + + # Give build_dag the repository assets that ``asimov manage build`` stores + # before submission. The coinc file is replaced from the bootstrap below, + # but its initial presence avoids any GraceDB access during this test. + repository_dir = Path(event.repository.directory) + if not repository_dir.is_absolute(): + repository_dir = project_dir / repository_dir + category_dir = repository_dir / production.category + category_dir.mkdir(parents=True, exist_ok=True) + (category_dir / "coinc.xml").write_text("synthetic coinc\n") + + commands = [] + + class SchedulerBoundary: + def __init__(self, command, **kwargs): + commands.append(command) + executable = Path(command[0]).name + if executable == "util_RIFT_pseudo_pipe.py": + rundir = Path(production.rundir) + rundir.mkdir(parents=True, exist_ok=True) + dag = ( + rundir + / "marginalize_intrinsic_parameters_BasicIterationWorkflow.dag" + ) + dag.write_text( + "# synthetic DAG emitted at the external RIFT boundary\n" + ) + self.stdout = b"RIFT DAG prepared" + elif executable == "condor_submit_dag": + self.stdout = b"submitted to cluster 4242." + else: + raise AssertionError("unexpected external command: {}".format(command)) + + def communicate(self): + return self.stdout, None + + # Run Rimsky's real Asimov submission helper. Only the heavyweight + # pseudo-pipeline process and final scheduler process are replaced; PSD + # conversion, config rendering, posterior reading, and bootstrap conversion + # run for real against the synthetic files above. + with chdir(Path(settings.asimovdir)), patch("asimov.git.time.sleep"), patch( + "RIFT.asimov.rift.subprocess.Popen", SchedulerBoundary + ): + build_and_submit(event, production, ledger) + + bootstrap = category_dir / "rift-online_bootstrap.xml.gz" + assert bootstrap.is_file() + assert (category_dir / "coinc.xml").is_file() + configuration = category_dir / "rift-online.ini" + assert configuration.is_file() + rendered = configuration.read_text() + assert "fake-cache" in rendered + assert str(Path(caches["H1"])) in rendered + assert production.status == "running" + assert production.job_id == 4242 + assert [Path(command[0]).name for command in commands] == [ + "util_RIFT_pseudo_pipe.py", + "condor_submit_dag", + ]