Added script to run the Global Hazard Mosaic tests

openquake/baselib/parallel.py

@@ -1109,14 +1109,23 @@ def multispawn(func, allargs, nprocs=num_cores, logfinish=True,
             func(*args)
         return
     tot = len(allargs)
+    allargs = allargs[::-1]  # so that the first argument is submitted first
     procs = {}  # sentinel -> process
     n = 1
     while allargs:
-        args = allargs.pop(0)
-        name = names.pop(0) if names else None
+        args = allargs.pop()
+        name = names.pop() if names else None
         proc = mp_context.Process(target=func, args=args, name=name)
         proc.start()
         procs[proc.sentinel] = proc
+        while len(procs) >= nprocs:  # wait for something to finish
+            for finished in wait(procs):
+                procs[finished].join()
+                del procs[finished]
+                if logfinish:
+                    logging.info('Finished job %s [%d of %d]', name, n, tot)
+                n += 1
+
     while procs:
         for finished in wait(procs):
             name = procs[finished].name or ''

openquake/calculators/base.py

@@ -303,7 +303,7 @@ def check_precalc(self, precalc_mode):
                 (calc_mode, ok_mode, precalc_mode))
 
     def run(self, pre_execute=True, concurrent_tasks=None, remove=False,
-            shutdown=False, **kw):
+            shutdown=False, name='JOB', **kw):
         """
         Run the calculation and return the exported outputs.
 
@@ -373,7 +373,9 @@ def run(self, pre_execute=True, concurrent_tasks=None, remove=False,
                         # removing in preclassical with multiFaultSources
                         # would break --hc which is reading the temp file
                         os.remove(self.datastore.tempname)
-        return getattr(self, 'exported', {})
+        exportdic = getattr(self, 'exported', {})
+        exportdic[name] = name
+        return exportdic
 
     def core_task(*args):
         """

openquake/engine/engine.py

@@ -218,7 +218,7 @@ def run_calc(log):
         workflow = oqparam.calculation_mode == 'workflow'
         set_concurrent_tasks_default(calc, 1 if workflow else 2)
         t0 = time.time()
-        calc.run(shutdown=True)
+        calc.run(shutdown=True, name=getattr(log, 'name', 'JOB{log.calc_id}'))
         logging.info('Exposing the outputs to the database')
         expose_outputs(calc.datastore)
         calc.datastore.close()
@@ -338,12 +338,11 @@ def _run(jobctxs, job_id, nodes, sbatch, concurrent_jobs, notify_to):
             w.WorkerMaster(job_id).send_jobs()
             print('oq engine --show-log %d to see the progress' % job_id)
         elif concurrent_jobs > 1:
-            args = [(job,) for job in jobctxs]
-            names = []
+            allargs = []
             for job in jobctxs:
-                name = f"{job.params['mosaic_model']}{job.calc_id}"
-                names.append(name)
-            parallel.multispawn(run_calc, args, concurrent_jobs, names=names)
+                job.name = f"{job.params['mosaic_model']}{job.calc_id}"
+                allargs.append((job,))
+            parallel.multispawn(run_calc, allargs, concurrent_jobs)
         else:
             for jobctx in jobctxs:
                 run_calc(jobctx)
@@ -535,9 +534,10 @@ def validate(self):
             oqs.append(oq)
         if 'classical' in oq.calculation_mode:
             for oq in oqs[1:]:
+                ini = oq.inputs['job_ini']
                 if oq.retperiods != oqs[0].retperiods:
-                    raise NameError(
-                        f'Expected return_periods = {oqs[0].retperiods}, '
+                    raise InvalidFile(
+                        f'{ini}: expected return_periods = {oqs[0].retperiods}, '
                         f'got {oq.retperiods}')
         if 'risk' in oq.calculation_mode or 'damage' in oq.calculation_mode:
             for oq in oqs[1:]:

openquake/engine/tests/global_hazard_test.py

@@ -0,0 +1,33 @@
+# -*- coding: utf-8 -*-
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Copyright (C) 2024-2025, GEM Foundation
+#
+# OpenQuake is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# OpenQuake is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with OpenQuake.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+from openquake.qa_tests_data import mosaic
+from openquake.commonlib.datastore import read
+from openquake.calculators.export import export
+from openquake.engine import global_hazard
+
+MOSAIC_DIR = os.path.dirname(mosaic.__file__)
+
+
+# TODO: enable the test
+def _test():
+    worflow_id = global_hazard.main(MOSAIC_DIR, 'job_vs30.ini')
+    dstore = read(worflow_id)
+    [fname] = export(('hmaps', 'csv'), dstore)
+    breakpoint()