diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/README.md b/MonteCarloMarginalizeCode/Code/RIFT/asimov/README.md index f2b805f46..9290ca086 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/README.md +++ b/MonteCarloMarginalizeCode/Code/RIFT/asimov/README.md @@ -7,3 +7,16 @@ 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 + +Compatibility notes +------------------- + +With ASIMOV versions that provide ``PESummaryPipeline``, RIFT retains the +legacy automatic PESummary completion job. ASIMOV 0.7 and newer manage +PESummary as a separate postprocessing analysis, so RIFT marks the PE analysis +finished and does not submit a duplicate postprocessing job. + +``Rift.collect_assets(absolute=True)`` publishes the ``rift-assets/v1`` +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. diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini index db21ac80d..d34121f81 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini +++ b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.ini @@ -68,7 +68,11 @@ types = { {% for ifo in ifos %}"{{ifo}}":"{{data['frame types'][ifo]}}",{% endfo channels = { {% for ifo in ifos %}"{{ifo}}":"{{data['channels'][ifo]}}",{% endfor %} } [lalinference] +{% if likelihood contains 'minimum frequency' %} +flow = { {% for ifo in ifos %}"{{ifo}}":{{likelihood['minimum frequency'][ifo]}},{% endfor %} } +{% else %} flow = { {% for ifo in ifos %}"{{ifo}}":{{quality['minimum frequency'][ifo]}},{% endfor %} } +{% endif %} fhigh = { {% for ifo in ifos %}"{{ifo}}":{{quality['maximum frequency'][ifo]}},{% endfor %} } [engine] diff --git a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py index 1fa57bd42..ba10cfef0 100644 --- a/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py +++ b/MonteCarloMarginalizeCode/Code/RIFT/asimov/rift.py @@ -12,7 +12,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 @@ -78,6 +83,18 @@ 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.7 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 # Top-level groups a PESummary metafile carries that are not analysis labels _PESUMMARY_RESERVED = ('version', 'history') @@ -192,10 +209,19 @@ def _find_posterior(self): for production in self.production.event.productions: productions[production.name] = production for previous_job in self.production.dependencies: - self.logger.info("RIFT: previous job assets" + str( productions[previous_job].pipeline.collect_assets())) try: - if "samples" in productions[previous_job].pipeline.collect_assets(): - posterior_file = productions[previous_job].pipeline.collect_assets()['samples'] + previous_assets = productions[previous_job].pipeline.collect_assets() + self.logger.info("RIFT: previous job assets" + str(previous_assets)) + if "samples" in previous_assets: + posterior_file = previous_assets['samples'] + if isinstance(posterior_file, (list, tuple)): + if len(posterior_file) != 1: + raise PipelineException( + "RIFT bootstrap: {} publishes {} sample files; " + "need exactly one PESummary metafile".format( + previous_job, len(posterior_file)), + production=self.production.name) + posterior_file = posterior_file[0] self.production.meta['dataset'] = self._dataset_label(posterior_file) return posterior_file except PipelineException: @@ -211,10 +237,36 @@ def _find_posterior(self): else: self.logger.error("Could not find an analysis providing posterior samples to analyse.") + def _reuse_existing_bootstrap(self, bootstrap_file, posterior_file): + """Fail closed unless reuse of an unprovenanced grid is explicit.""" + if not os.path.exists(bootstrap_file): + return False + if not self.production.meta['scheduler'].get( + 'bootstrap reuse existing', False): + raise PipelineException( + "RIFT bootstrap: existing grid {} may come from a different " + "posterior than {}. Remove the grid, use a new analysis name, " + "or explicitly set scheduler: bootstrap reuse existing: true." + .format(bootstrap_file, posterior_file), + production=self.production.name) + self.logger.warning( + "RIFT bootstrap: explicitly reusing existing grid {} without " + "source provenance validation".format(bootstrap_file)) + return True + 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) @@ -248,7 +300,7 @@ def before_config(self, dryrun=False): category = config.get("general", "calibration_directory") # 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}"): @@ -489,14 +541,8 @@ def build_dag(self, user=None, dryrun=False): ) bootstrap_file_ascii = str(bootstrap_file) + "_ascii" # test if bootstrap file already exists - if os.path.exists(bootstrap_file): - # Rebuilding an analysis under the same name reuses this - # silently, so a changed bootstrap source has no effect. - self.logger.warning( - "RIFT bootstrap: reusing existing grid {} and IGNORING {}; " - "delete it (and its _ascii) to rebuild".format( - bootstrap_file, posterior_file)) - if not(os.path.exists(bootstrap_file)): + if not self._reuse_existing_bootstrap( + bootstrap_file, posterior_file): import RIFT.misc.samples_utils RIFT.misc.samples_utils.dump_pesummary_samples_to_file_as_rift(posterior_file, self.production.meta['dataset'], bootstrap_file_ascii) extra_args ='' @@ -625,7 +671,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 +714,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 +725,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]}" ) @@ -844,7 +890,11 @@ def detect_completion(self): def collect_assets(self,absolute=False): """ - Gather all of the results assets for this job. + Gather result assets for downstream ASIMOV/PESummary analyses. + + ``samples`` is always a list, including calibration-reweighted output. + Consumers which run outside the RIFT working directory should request + absolute paths. """ if absolute: rundir = os.path.abspath(self.production.rundir) @@ -852,11 +902,58 @@ def collect_assets(self,absolute=False): rundir = self.production.rundir rift_all_lnL = os.path.join(rundir, 'all.net') samples_raw = os.path.join(rundir,'extrinsic_posterior_samples.dat') - dict_out = {"samples":self.samples(), "lnL_marg":rift_all_lnL, "samples_raw":samples_raw} + dict_out = { + "samples": self.samples(absolute=absolute), + "lnL_marg": rift_all_lnL, + "samples_raw": samples_raw, + "provenance": { + "pipeline": "rift", + "event": self.production.event.name, + "analysis": self.production.name, + }, + } rewt_file_name = os.path.join(rundir,'reweighted_posterior_samples.dat') if os.path.exists(rewt_file_name): dict_out['samples_calmarg'] = rewt_file_name - dict_out['samples'] = rewt_file_name + dict_out['samples'] = [rewt_file_name] + + try: + ini = self.production.get_configuration().ini_loc + if not os.path.isabs(ini): + ini = os.path.join( + self.production.event.repository.directory, + self.category, + ini, + ) + dict_out["config"] = os.path.abspath(ini) if absolute else ini + except (AttributeError, IndexError, TypeError, ValueError): + self.logger.warning("RIFT configuration asset is not available") + + psds = self._get_psds("ascii") + if psds: + dict_out["psds"] = { + ifo: (os.path.abspath(path) if os.path.isabs(path) else + os.path.abspath(os.path.join( + self.production.event.repository.directory, path))) + if absolute else path + for ifo, path in psds.items() + } + + calibration = self.production.meta.get("data", {}).get("calibration", {}) + if calibration: + dict_out["calibration"] = { + ifo: (os.path.abspath(path) if os.path.isabs(path) else + os.path.abspath(os.path.join( + self.production.event.repository.directory, path))) + if absolute else path + for ifo, path in calibration.items() + } + + if dict_out["samples"] and "config" in dict_out: + dict_out["asset_contract"] = "rift-assets/v1" + else: + self.logger.warning( + "RIFT assets are incomplete; not advertising rift-assets/v1") return dict_out diff --git a/MonteCarloMarginalizeCode/Code/test/test_asimov_compatibility.py b/MonteCarloMarginalizeCode/Code/test/test_asimov_compatibility.py new file mode 100644 index 000000000..48481f125 --- /dev/null +++ b/MonteCarloMarginalizeCode/Code/test/test_asimov_compatibility.py @@ -0,0 +1,277 @@ +#!/usr/bin/env python3 +"""Cross-version contract tests for the RIFT ASIMOV adapter.""" + +import os +import types + +import pytest + +pytest.importorskip("asimov") +rift_asimov = pytest.importorskip("RIFT.asimov.rift") + +Rift = rift_asimov.Rift +PipelineException = rift_asimov.PipelineException + + +class _Logger: + def info(self, *args, **kwargs): + pass + + def warning(self, *args, **kwargs): + pass + + +def _pipe(production): + pipe = Rift.__new__(Rift) + pipe.production = production + pipe.category = production.category + pipe.logger = _Logger() + return pipe + + +def test_asimov_07_completion_defers_to_separate_postprocessing(monkeypatch): + production = types.SimpleNamespace( + status="processing", category="C01_offline", meta={"job id": 12} + ) + pipe = _pipe(production) + monkeypatch.setattr(rift_asimov, "PESummaryPipeline", None) + + pipe.after_completion() + + assert production.status == "finished" + + +def test_legacy_completion_submits_pesummary_once(monkeypatch): + calls = [] + + class _LegacyPESummary: + def __init__(self, production, category=None): + calls.append((production, category)) + + def submit_dag(self): + return 314 + + production = types.SimpleNamespace( + status="running", category="C01_offline", meta={} + ) + pipe = _pipe(production) + monkeypatch.setattr(rift_asimov, "PESummaryPipeline", _LegacyPESummary) + + pipe.after_completion() + + assert calls == [(production, "C01_offline")] + assert production.meta["job id"] == 314 + assert production.status == "processing" + + +def test_collect_assets_publishes_pesummary_inputs(tmp_path): + rundir = tmp_path / "run" + rundir.mkdir() + samples = rundir / "extrinsic_posterior_samples.dat" + samples.write_text("# samples\n") + config = tmp_path / "repository" / "C01_offline" / "rift.ini" + config.parent.mkdir(parents=True) + config.write_text("[analysis]\n") + psd = tmp_path / "H1-psd.dat" + psd.write_text("20 1e-46\n") + calibration = tmp_path / "H1-calibration.dat" + calibration.write_text("20 0 0\n") + + repository = types.SimpleNamespace(directory=str(tmp_path / "repository")) + event = types.SimpleNamespace(name="S250202cu", repository=repository) + production = types.SimpleNamespace( + name="rift-SEOBNRv5PHM", + category="C01_offline", + rundir=str(rundir), + event=event, + psds={"H1": str(psd)}, + xml_psds={}, + meta={"data": {"calibration": {"H1": str(calibration)}}}, + get_configuration=lambda: types.SimpleNamespace(ini_loc="rift.ini"), + ) + + assets = _pipe(production).collect_assets(absolute=True) + + assert assets["asset_contract"] == "rift-assets/v1" + assert assets["samples"] == [str(samples)] + assert assets["config"] == str(config) + assert assets["psds"] == {"H1": str(psd)} + assert assets["calibration"] == {"H1": str(calibration)} + assert assets["provenance"] == { + "pipeline": "rift", + "event": "S250202cu", + "analysis": "rift-SEOBNRv5PHM", + } + + +def test_reweighted_samples_keep_list_contract(tmp_path): + rundir = tmp_path / "run" + rundir.mkdir() + reweighted = rundir / "reweighted_posterior_samples.dat" + reweighted.write_text("# samples\n") + event = types.SimpleNamespace( + name="S250202cu", repository=types.SimpleNamespace(directory=str(tmp_path)) + ) + production = types.SimpleNamespace( + name="rift-calmarg", + category="C01_offline", + rundir=str(rundir), + event=event, + psds={}, + meta={"data": {}}, + get_configuration=lambda: (_ for _ in ()).throw(ValueError()), + ) + + assets = _pipe(production).collect_assets(absolute=True) + + assert assets["samples"] == [str(reweighted)] + assert assets["samples_calmarg"] == str(reweighted) + assert "asset_contract" not in assets + + +def test_collect_assets_resolves_relative_detector_paths_from_repository( + tmp_path, monkeypatch): + repository_dir = tmp_path / "repository" + run = tmp_path / "run" + run.mkdir() + (run / "extrinsic_posterior_samples.dat").write_text("# samples\n") + config = repository_dir / "C01_offline" / "rift.ini" + config.parent.mkdir(parents=True) + config.write_text("[analysis]\n") + psd = repository_dir / "assets" / "H1-psd.dat" + calibration = repository_dir / "assets" / "H1-calibration.dat" + psd.parent.mkdir() + psd.write_text("20 1e-46\n") + calibration.write_text("20 0 0\n") + elsewhere = tmp_path / "elsewhere" + elsewhere.mkdir() + monkeypatch.chdir(elsewhere) + + event = types.SimpleNamespace( + name="S250202cu", + repository=types.SimpleNamespace(directory=str(repository_dir)), + ) + production = types.SimpleNamespace( + name="rift-relative", + category="C01_offline", + rundir=str(run), + event=event, + psds={"H1": "assets/H1-psd.dat"}, + xml_psds={}, + meta={"data": {"calibration": { + "H1": "assets/H1-calibration.dat"}}}, + get_configuration=lambda: types.SimpleNamespace(ini_loc="rift.ini"), + ) + + assets = _pipe(production).collect_assets(absolute=True) + + assert assets["psds"] == {"H1": str(psd)} + assert assets["calibration"] == {"H1": str(calibration)} + assert assets["asset_contract"] == "rift-assets/v1" + + +def test_collect_assets_distinguishes_standard_calmarg_and_all_net(tmp_path): + run = tmp_path / "run" + run.mkdir() + standard = run / "extrinsic_posterior_samples.dat" + calmarg = run / "reweighted_posterior_samples.dat" + all_net = run / "all.net" + standard.write_text("# standard\n") + calmarg.write_text("# calmarg\n") + all_net.write_text("# likelihood\n") + repository = tmp_path / "repository" + config = repository / "C01_offline" / "rift.ini" + config.parent.mkdir(parents=True) + config.write_text("[analysis]\n") + event = types.SimpleNamespace( + name="S250202cu", + repository=types.SimpleNamespace(directory=str(repository)), + ) + production = types.SimpleNamespace( + name="rift-both", category="C01_offline", rundir=str(run), + event=event, psds={}, xml_psds={}, meta={"data": {}}, + get_configuration=lambda: types.SimpleNamespace(ini_loc="rift.ini"), + ) + + assets = _pipe(production).collect_assets(absolute=True) + + assert assets["samples"] == [str(calmarg)] + assert assets["samples_raw"] == str(standard) + assert assets["samples_calmarg"] == str(calmarg) + assert assets["lnL_marg"] == str(all_net) + assert assets["asset_contract"] == "rift-assets/v1" + + +def test_asimov_07_psd_attributes_replace_legacy_getter(): + production = types.SimpleNamespace( + category="C01_offline", + psds={"H1": "/tmp/H1.dat"}, + xml_psds={"H1": "/tmp/H1.xml.gz"}, + ) + pipe = _pipe(production) + + assert pipe._get_psds("ascii") == production.psds + assert pipe._get_psds("xml") == ["/tmp/H1.xml.gz"] + + +def test_single_sample_list_is_unwrapped_for_bootstrap(monkeypatch): + dependency = types.SimpleNamespace( + name="pesummary", + pipeline=types.SimpleNamespace( + collect_assets=lambda: {"samples": ["combined.h5"]} + ), + ) + event = types.SimpleNamespace(productions=[dependency]) + production = types.SimpleNamespace( + name="rift-bootstrap", + category="C01_offline", + dependencies=["pesummary"], + event=event, + meta={"scheduler": {}}, + ) + pipe = _pipe(production) + monkeypatch.setattr(pipe, "_dataset_label", lambda path: "rift-source") + + assert pipe._find_posterior() == "combined.h5" + assert production.meta["dataset"] == "rift-source" + + +def test_multiple_sample_files_are_rejected_for_bootstrap(): + dependency = types.SimpleNamespace( + name="pesummary", + pipeline=types.SimpleNamespace( + collect_assets=lambda: {"samples": ["a.h5", "b.h5"]} + ), + ) + event = types.SimpleNamespace(productions=[dependency]) + production = types.SimpleNamespace( + name="rift-bootstrap", + category="C01_offline", + dependencies=["pesummary"], + event=event, + meta={"scheduler": {}}, + ) + + with pytest.raises(PipelineException, match="exactly one PESummary metafile"): + _pipe(production)._find_posterior() + + +def test_existing_bootstrap_requires_explicit_unprovenanced_reuse(tmp_path): + bootstrap = tmp_path / "bootstrap.xml.gz" + bootstrap.write_text("old grid") + production = types.SimpleNamespace( + name="rift-bootstrap", category="C01_offline", + meta={"scheduler": {}}, + ) + pipe = _pipe(production) + + with pytest.raises(PipelineException, match="bootstrap reuse existing"): + pipe._reuse_existing_bootstrap(str(bootstrap), "new-posterior.h5") + + production.meta["scheduler"]["bootstrap reuse existing"] = True + assert pipe._reuse_existing_bootstrap( + str(bootstrap), "new-posterior.h5") is True + + +if __name__ == "__main__": + raise SystemExit(pytest.main([os.path.abspath(__file__), "-v"]))