diff --git a/.github/workflows/rimsky-integration.yml b/.github/workflows/rimsky-integration.yml new file mode 100644 index 000000000..61c162cc6 --- /dev/null +++ b/.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 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 9290ca086..e1e795d6c 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/README.md +++ b/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. diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini index d34121f81..ccc3a225a 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini +++ b/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" diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py index ba10cfef0..be2ff3f2f 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 @@ -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 = [ diff --git a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/README.md new file mode 100644 index 000000000..b29a97f64 --- /dev/null +++ b/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. diff --git a/MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py new file mode 100644 index 000000000..6020ad1de --- /dev/null +++ b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/__init__.py @@ -0,0 +1,27 @@ +"""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, + 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 new file mode 100644 index 000000000..c45a41765 --- /dev/null +++ b/MonteCarloMarginalizeCode/Code/RIFT/rimsky/integration.py @@ -0,0 +1,311 @@ +"""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 _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. + + 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", + "orchestrator": "rimsky", + "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 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. + + 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 write_rimsky_config(config, path): + """Write a Rimsky configuration containing the automatic RIFT hook.""" + return write_analysis(config, path) + + +def main(argv=None): + """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) + analysis_path = write_analysis( + build_analysis(config, config_path=args.rimsky_config), args.output + ) + 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__": + 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..897b2e28f --- /dev/null +++ b/MonteCarloMarginalizeCode/Code/test/test_rimsky_end_to_end.py @@ -0,0 +1,291 @@ +"""Submission-level test for Rimsky's real Asimov follow-up hook. + +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 + +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, build_and_submit + +import RIFT.asimov.rift as rift_asimov +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 = {environment} + +[condor] +user = rimsky-test + +[general] +git_default = . +rundir_default = {project}/working +calibration = test +calibration_directory = C01_offline +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, 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": {}}}) + asimov.current_ledger = ledger + return ledger + + +def test_first_rimsky_result_creates_bootstrapped_rift_production( + tmp_path, monkeypatch +): + assert Version(version("asimov")) >= Version("0.7") + + 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) + 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 = {} + 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) + np.savetxt(psd, [[0, 1e-40], [1, 1e-40], [2, 1e-40]]) + 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()) + + # 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", + ] diff --git a/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py b/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py new file mode 100644 index 000000000..597f65267 --- /dev/null +++ b/MonteCarloMarginalizeCode/Code/test/test_rimsky_integration.py @@ -0,0 +1,298 @@ +import copy +import configparser +from pathlib import Path +from types import SimpleNamespace + +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, +) + + +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["orchestrator"] == "rimsky" + 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={ + "orchestrator": "rimsky", + "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_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" + ).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 + + +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) + 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 557634b81..bdcdcc576 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':