Add ASIMOV 0.7 compatibility and PESummary assets

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.

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]

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,19 +890,70 @@ 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)
         else:
             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