CHANGES.rst

@@ -4,7 +4,15 @@ Changelog
 1.0.2 (unreleased)
 ------------------
 
--   No changes yet
+-   Remove gstlal from list of pipelines that upload PSDs in separate
+    ``psd.xml.gz`` file. Update mock event uploads to include PSD in
+    ``coinc.xml`` upload.
+-   Disable lalinference parameter estimation and switch to bilby as a main
+    parameter estimation software.
+-   Add bilby-pipe>=1.0.6 and gwdatafind>=1.1.1 dependency, and unpin pesummary
+    to fix bilby workflow for O3-replay.
+-   Disable parameter estimation for MDC events as it is currently broken for
+    them.
 
 1.0.1 (2022-05-09)
 ------------------

gwcelery/tasks/first2years.py

@@ -137,12 +137,6 @@ def _vet_event(superevents):
         ).apply_async()
 
 
-@gracedb.task(ignore_result=True, shared=False)
-def _upload_psd(graceid):
-    psd = resources.read_binary(data_first2years, 'psd.xml.gz')
-    gracedb.upload(psd, 'psd.xml.gz', graceid, 'Noise PSD', ['psd'])
-
-
 @app.on_after_finalize.connect
 def setup_periodic_tasks(sender, **kwargs):
     """Register periodic tasks.
@@ -168,17 +162,11 @@ def upload_event():
     if app.conf['mock_events_simulate_multiple_uploads']:
         num = 15
         for _ in range(num):
-            (
-                gracedb.create_event.s(
-                    _jitter_snr(coinc), 'MDC', 'gstlal', 'CBC'
-                )
-                |
-                _upload_psd.s()
+            gracedb.create_event.s(
+                 _jitter_snr(coinc), 'MDC', 'gstlal', 'CBC'
             ).apply_async()
 
     (
-        _upload_psd.si(graceid)
-        |
         gracedb.get_superevents.si(
             'MDC event: {}'.format(graceid)
         ).set(countdown=600)

gwcelery/tasks/inference.py

@@ -268,13 +268,16 @@ def _setup_dag_for_lalinference(coinc_psd, ini_contents,
 
 
 @app.task(shared=False)
-def _setup_dag_for_bilby(event, rundir, preferred_event_id, superevent_id):
+def _setup_dag_for_bilby(
+    event_coinc, rundir, preferred_event_id, superevent_id
+):
     """Create DAG for a bilby run and return the path to DAG.
 
     Parameters
     ----------
-    event : json contents
-        The json contents retrieved from gracedb.get_event()
+    event_coinc : tuple
+        Tuple of the json contents retrieved from gracedb.get_event() and
+        the byte contents of coinc.xml
     rundir : str
         The path to a run directory where the DAG file exits
     preferred_event_id : str
@@ -288,20 +291,26 @@ def _setup_dag_for_bilby(event, rundir, preferred_event_id, superevent_id):
         The path to the .dag file
 
     """
+    event, coinc = event_coinc
+
     path_to_json = os.path.join(rundir, 'event.json')
     with open(path_to_json, 'w') as f:
         json.dump(event, f, indent=2)
 
+    path_to_coinc = os.path.join(rundir, 'coinc.xml')
+    with open(path_to_coinc, 'wb') as f:
+        f.write(coinc)
+
     path_to_webdir = os.path.join(
         app.conf['pe_results_path'], preferred_event_id, 'bilby'
     )
 
     setup_arg = ['bilby_pipe_gracedb', '--webdir', path_to_webdir,
-                 '--outdir', rundir, '--json', path_to_json, '--online-pe',
-                 '--convert-to-flat-in-component-mass']
+                 '--outdir', rundir, '--json', path_to_json,
+                 '--psd-file', path_to_coinc, '--online-pe']
 
     if not app.conf['gracedb_host'] == 'gracedb.ligo.org':
-        setup_arg += ['--channel-dict', 'o2replay',
+        setup_arg += ['--channel-dict', 'o3replay',
                       '--sampler-kwargs', 'FastTest']
     try:
         subprocess.run(setup_arg, capture_output=True, check=True)
@@ -323,7 +332,6 @@ def _setup_dag_for_bilby(event, rundir, preferred_event_id, superevent_id):
             'pe', 'online_bilby_pe.ini')).delay()
 
     path_to_dag, = glob.glob(os.path.join(rundir, 'submit/dag*.submit'))
-    print(path_to_dag)
     return path_to_dag
 
 
@@ -367,8 +375,10 @@ def dag_prepare_task(rundir, superevent_id, preferred_event_id, pe_pipeline,
             _download_psd.si(preferred_event_id)
         ) | _setup_dag_for_lalinference.s(ini_contents, rundir, superevent_id)
     elif pe_pipeline == 'bilby':
-        canvas = gracedb.get_event.si(preferred_event_id) | \
-            _setup_dag_for_bilby.s(rundir, preferred_event_id, superevent_id)
+        canvas = group(
+            gracedb.get_event.si(preferred_event_id),
+            gracedb.download.si('coinc.xml', preferred_event_id)
+        ) | _setup_dag_for_bilby.s(rundir, preferred_event_id, superevent_id)
     else:
         raise NotImplementedError(f'Unknown PE pipeline {pe_pipeline}.')
     canvas |= _condor_no_submit.s()
@@ -572,10 +582,16 @@ def dag_finished(rundir, preferred_event_id, superevent_id, pe_pipeline):
         ]
     elif pe_pipeline == 'bilby':
         resultdir = os.path.join(rundir, 'result')
+        sampledir = os.path.join(rundir, 'final_result')
+        sample_filename = 'Bilby.posterior_samples.hdf5'
+        subprocess.run(
+            ['bilby_pipe_to_ligo_skymap_samples',
+             os.path.join(sampledir, '*result.hdf5'),
+             '--out', os.path.join(sampledir, sample_filename)])
         uploads = [
-            (resultdir, '*merge_result.json',
+            (sampledir, sample_filename,
              'Bilby posterior samples',
-             'Bilby.posterior_samples.json'),
+             sample_filename),
             (resultdir, '*_extrinsic_corner.png',
              'Bilby corner plot for extrinsic parameters',
              'Bilby.extrinsic.png'),
@@ -601,7 +617,7 @@ def dag_finished(rundir, preferred_event_id, superevent_id, pe_pipeline):
         clean_up.si(rundir)
     ).delay()
 
-    if pe_pipeline == 'lalinference':
+    if pe_pipeline == 'bilby':
         gracedb.create_label.delay('PE_READY', superevent_id)
 
 

gwcelery/tasks/orchestrator.py

@@ -602,50 +602,33 @@ def _get_lowest_far(superevent_id):
 
 @app.task(ignore_result=True, shared=False)
 def parameter_estimation(far_event, superevent_id):
-    """Tasks for Parameter Estimation Followup with LALInference or Bilby
-
-    For LALInference, this consists of the following steps:
-
-    1.   Prepare and upload an ini file which is suitable for the target event.
-    2.   Start Parameter Estimation if FAR is smaller than the PE threshold.
-
-    For Bilby, this consists of the following steps:
+    """Parameter Estimation with Bilby. This consists of the following steps:
 
     1.   Start Parameter Estimation if FAR is smaller than the PE threshold.
     2.   Upload of ini file during Parameter Estimation
     """
     far, event = far_event
+    # TODO: parameter estimation is disabled for mock uploads as it is
+    # currently broken for them.
+    if event['group'] != 'CBC' or event['search'] == 'MDC':
+        return
     preferred_event_id = event['graceid']
     threshold = (app.conf['preliminary_alert_far_threshold']['cbc'] /
                  app.conf['preliminary_alert_trials_factor']['cbc'])
-    # FIXME: it will be better to start parameter estimation for 'burst'
-    # events.
-    is_production = (app.conf['gracedb_host'] == 'gracedb.ligo.org')
-    is_mdc = (event['search'] == 'MDC')
-    if event['group'] == 'CBC' and not (is_production and is_mdc):
-        canvas = inference.pre_pe_tasks(event, superevent_id)
-        if far <= threshold:
-            pipelines = ['lalinference']
-            # FIXME: The second condition guarantees that the bilby for
-            # playground or test events are started less than once per day to
-            # save computational resources. Once bilby becomes quick enough, we
-            # should drop that condition.
-            if is_production or (is_mdc and superevent_id[8:] == 'a'):
-                pipelines.append('bilby')
-            canvas |= group(
-                inference.start_pe.s(preferred_event_id, superevent_id, p)
-                for p in pipelines)
-        else:
-            canvas |= gracedb.upload.si(
-                          filecontents=None, filename=None,
-                          graceid=superevent_id,
-                          message='FAR is larger than the PE threshold, '
-                                  '{}  Hz. Parameter Estimation will not '
-                                  'start.'.format(threshold),
-                          tags='pe'
-                      )
-
-        canvas.apply_async()
+    canvas = inference.pre_pe_tasks(event, superevent_id)
+    if far <= threshold:
+        canvas |= inference.start_pe.s(
+            preferred_event_id, superevent_id, 'bilby'
+        )
+    else:
+        canvas |= gracedb.upload.si(
+            filecontents=None, filename=None,
+            graceid=superevent_id,
+            message='FAR is larger than the PE threshold, {}  Hz. '
+                    'Parameter Estimation will not start.'.format(threshold),
+            tags='pe'
+        )
+    canvas.apply_async()
 
 
 @gracedb.task(ignore_result=True, shared=False)

gwcelery/tests/test_tasks_first2years.py

-from importlib import resources
 import io
 from unittest.mock import call, patch
 
@@ -8,7 +7,6 @@ from ligo.skymap.io.events.ligolw import ContentHandler
 import pytest
 
 from ..tasks.first2years import pick_coinc, upload_event
-from ..data import first2years as data_first2years
 
 pytest.importorskip('lal')
 
@@ -33,21 +31,16 @@ def test_pick_coinc():
 
 @patch('lal.GPSTimeNow', mock_now)
 @patch('gwcelery.tasks.gracedb.create_event', return_value='M1234')
-@patch('gwcelery.tasks.gracedb.upload.run')
 @patch('gwcelery.tasks.gracedb.get_superevents.run',
        return_value=[{'superevent_id': 'S1234'}])
 @patch('gwcelery.tasks.gracedb.create_signoff.run')
 def test_upload_event(mock_create_signoff, mock_get_superevents,
-                      mock_upload, mock_create_event):
+                      mock_create_event):
     coinc = pick_coinc()
-    psd = resources.read_binary(data_first2years, 'psd.xml.gz')
 
     upload_event()
 
     mock_create_event.assert_called_once_with(coinc, 'MDC', 'gstlal', 'CBC')
-    mock_upload.assert_has_calls([
-        call(psd, 'psd.xml.gz', 'M1234', 'Noise PSD', ['psd'])
-    ])
     mock_get_superevents.assert_called_once_with('MDC event: M1234')
     mock_create_signoff.assert_called_once()
     msg = ('If this had been a real gravitational-wave event candidate, '

gwcelery/tests/test_tasks_inference.py

@@ -197,13 +197,13 @@ def test_setup_dag_for_lalinference_(monkeypatch, tmp_path, psd):
 @pytest.mark.parametrize(
     'host', ['gracedb-playground.ligo.org', 'gracedb.ligo.org'])
 def test_setup_dag_for_bilby(monkeypatch, tmp_path, host):
-    event, rundir = {}, str(tmp_path)
+    event_coinc, rundir = ({}, b'coinc'), str(tmp_path)
     pid, sid = 'G1234', 'S1234'
     monkeypatch.setitem(app.conf, 'gracedb_host', host)
     dag = 'bilby dag'
 
     def _subprocess_run(cmd, **kwargs):
-        is_quick_conf = 'o2replay' in cmd and 'FastTest' in cmd
+        is_quick_conf = 'o3replay' in cmd and 'FastTest' in cmd
         if host == 'gracedb.ligo.org':
             assert not is_quick_conf
         else:
@@ -219,7 +219,7 @@ def test_setup_dag_for_bilby(monkeypatch, tmp_path, host):
     monkeypatch.setattr('subprocess.run', _subprocess_run)
     monkeypatch.setattr('gwcelery.tasks.gracedb.upload.run', upload)
 
-    path_to_dag = inference._setup_dag_for_bilby(event, rundir, pid, sid)
+    path_to_dag = inference._setup_dag_for_bilby(event_coinc, rundir, pid, sid)
     with open(path_to_dag, 'r') as f:
         assert f.read() == dag
     upload.assert_called_once()
@@ -240,7 +240,8 @@ def test_setup_dag_for_failure(monkeypatch, tmp_path, pipeline):
             inference._setup_dag_for_lalinference(
                 (b'coinc', b'psd'), 'ini', rundir, 'S1234')
         elif pipeline == 'bilby':
-            inference._setup_dag_for_bilby({}, rundir, 'G1234', 'S1234')
+            inference._setup_dag_for_bilby(
+                ({}, b'coinc'), rundir, 'G1234', 'S1234')
     assert not os.path.exists(rundir)
     upload.assert_called_once()
 
@@ -265,8 +266,10 @@ def test_dag_prepare_task(monkeypatch, pipeline):
                 and r == rundir and s == sid)
         return path_to_dag
 
-    def _setup_dag_for_bilby(e, r, p, s):
-        assert e == event and r == rundir and p == gid and s == sid
+    def _setup_dag_for_bilby(e_c, r, p, s):
+        e, c = e_c
+        assert (e == event and c == coinc and r == rundir
+                and p == gid and s == sid)
         return path_to_dag
 
     def _subprocess_run(cmd, **kwargs):
@@ -335,11 +338,13 @@ def test_dag_finished(monkeypatch, tmp_path, pipeline):
     gid = 'G1234'
     rundir = str(tmp_path / 'rundir')
     resultdir = str(tmp_path / 'rundir/result')
+    sampledir = str(tmp_path / 'rundir/final_result')
     pe_results_path = str(tmp_path / 'public_html/online_pe')
     monkeypatch.setitem(app.conf, 'pe_results_path', pe_results_path)
     pe_results_path = os.path.join(pe_results_path, gid, pipeline)
     os.makedirs(rundir)
     os.makedirs(resultdir)
+    os.makedirs(sampledir)
     os.makedirs(pe_results_path)
 
     upload = Mock()
@@ -350,6 +355,7 @@ def test_dag_finished(monkeypatch, tmp_path, pipeline):
                         _upload_url)
     monkeypatch.setattr('gwcelery.tasks.gracedb.create_label.run',
                         create_label)
+    monkeypatch.setattr('subprocess.run', Mock())
 
     if pipeline in ['lalinference', 'bilby']:
         if pipeline == 'lalinference':
@@ -362,7 +368,7 @@ def test_dag_finished(monkeypatch, tmp_path, pipeline):
                      os.path.join(pe_results_path, 'extrinsic.png'),
                      os.path.join(pe_results_path, 'sourceFrame.png')]
         else:
-            paths = [os.path.join(resultdir, 'bilby_merge_result.json'),
+            paths = [os.path.join(sampledir, 'Bilby.posterior_samples.hdf5'),
                      os.path.join(resultdir, 'bilby_extrinsic_corner.png'),
                      os.path.join(resultdir, 'bilby_intrinsic_corner.png')]
         for path in paths:
@@ -372,7 +378,7 @@ def test_dag_finished(monkeypatch, tmp_path, pipeline):
         inference.dag_finished(rundir, gid, 'S1234', pipeline)
         assert upload.call_count == len(paths)
         _upload_url.assert_called_once()
-        if pipeline == 'lalinference':
+        if pipeline == 'bilby':
             create_label.assert_called_once()
         else:
             create_label.assert_not_called()

pyproject.toml

@@ -47,7 +47,7 @@ include = [
 [tool.poetry.dependencies]
 python = "^3.8"
 astropy = ">=4.3.1"  # https://github.com/astropy/astropy/issues/11879
-bilby-pipe = ">=0.3.8"
+bilby-pipe = ">=1.0.6"
 celery = {version = ">=5.1", extras = ["redis"]}
 click = "^7"
 comet = "*"
@@ -56,7 +56,7 @@ dnspython = "*"  # silence "DNS: dnspython not found. Can not use SRV lookup." w
 flask = "*"
 flask-caching = "*"
 gracedb-sdk = ">=0.1.5"
-gwdatafind = "*"
+gwdatafind = ">=1.1.1"
 gwpy = ">=2.0.1"  # https://github.com/gwpy/gwpy/issues/1277
 healpy = "*"
 igwn-alert = "*"
@@ -74,7 +74,7 @@ lscsoft-glue = "*"
 lxml = "*"
 numpy = "*"
 p-astro = ">=1.0"
-pesummary = "<0.13"  # https://git.ligo.org/lscsoft/pesummary/-/issues/269. FIXME: remove this requirement entirely once resolved.
+pesummary = "*"
 pygcn = ">=1.0.1"
 python-ligo-lw = "~1.7.0"  # https://git.ligo.org/lscsoft/ligo.skymap/-/merge_requests/255
 safe-netrc = "*"