From e29057d4f9248f2db8889ff94c55622fd0f8d41d Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Sun, 31 May 2026 05:25:12 +0200 Subject: [PATCH 01/15] Added global_hazard script --- openquake/engine/engine.py | 5 +- openquake/engine/global_hazard.py | 45 +++++++++++ .../AELO/aeloy3/py/Run_Jobs/CEA/gmmLT.xml | 77 ------------------- .../CEA/site10/job_dos_vs30_760_small.ini | 48 ------------ .../py/Run_Jobs/CEA/site10/sites_CEA_dos.csv | 2 - openquake/qa_tests_data/mosaic/workflow.py | 9 ++- 6 files changed, 54 insertions(+), 132 deletions(-) create mode 100755 openquake/engine/global_hazard.py delete mode 100644 openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/gmmLT.xml delete mode 100644 openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/job_dos_vs30_760_small.ini delete mode 100644 openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/sites_CEA_dos.csv diff --git a/openquake/engine/engine.py b/openquake/engine/engine.py index a8078d1d8d35..f7c9fa13cd56 100644 --- a/openquake/engine/engine.py +++ b/openquake/engine/engine.py @@ -538,9 +538,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:]: diff --git a/openquake/engine/global_hazard.py b/openquake/engine/global_hazard.py new file mode 100755 index 000000000000..09dd3c5953dc --- /dev/null +++ b/openquake/engine/global_hazard.py @@ -0,0 +1,45 @@ +#!/bin/env python +# -*- coding: utf-8 -*- +# vim: tabstop=4 shiftwidth=4 softtabstop=4 +# +# Copyright (C) 2026, 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 . + +import os +from unittest.mock import patch +from openquake.baselib import sap, config +from openquake.qa_tests_data.mosaic import workflow +from openquake.engine import engine + +def main(mosaic_dir, models='ALL', toml:bool=False, cache:str='false'): + """ + Storing global SES + """ + ghm_toml = workflow.ghm(mosaic_dir, models.split(','), toml) + if toml: + print(ghm_toml) + return + with patch.dict(config.directory, {'mosaic_dir': mosaic_dir}): + calc_id = engine.run_workflow(ghm_toml, {'cache': cache}) + os.remove(ghm_toml) + return calc_id + +main.mosaic_dir = 'Directory containing the hazard mosaic' +main.models = 'Models to consider (comma-separated)' +main.cache = 'Use the cache to avoid repeating correct calculations' +main.toml = 'Just print the TOML code' + +if __name__ == '__main__': + sap.run(main) diff --git a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/gmmLT.xml b/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/gmmLT.xml deleted file mode 100644 index a8a5aa379845..000000000000 --- a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/gmmLT.xml +++ /dev/null @@ -1,77 +0,0 @@ - - - - - - AbrahamsonEtAl2014 - 0.20000 - - - BooreEtAl2014 - 0.20000 - - - CampbellBozorgnia2014 - 0.20000 - - - ChiouYoungs2014 - 0.20000 - - - CauzziEtAl2014 - 0.20000 - - - - - NGAEastUSGSSeedFrankel - 0.12500 - - - NGAEastUSGSSeedHA15 - 0.12500 - - - NGAEastUSGSSeedSP15 - 0.12500 - - - NGAEastUSGSSeedYA15 - 0.12500 - - - NGAEastUSGSSammons1 - 0.12500 - - - NGAEastUSGSSammons2 - 0.12500 - - - NGAEastUSGSSammons3 - 0.12500 - - - NGAEastUSGSSammons4 - 0.12500 - - - - - AbrahamsonGulerce2020SSlab - 0.33000 - - - ParkerEtAl2020SSlab - 0.33000 - - - KuehnEtAl2020SSlab - 0.34000 - - - - - - diff --git a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/job_dos_vs30_760_small.ini b/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/job_dos_vs30_760_small.ini deleted file mode 100644 index 9471c6b3cc17..000000000000 --- a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/job_dos_vs30_760_small.ini +++ /dev/null @@ -1,48 +0,0 @@ -[general] -description = AELO Year 3 dos sites, CEA -calculation_mode = classical -random_seed = 23 -mosaic_model = CEA -ps_grid_spacing = 0 - -[geometry] -site_model_file = sites_CEA_dos.csv -sites = 76.953324 43.234055 -override_vs30 = 760 - -[logic_tree] -number_of_logic_tree_samples = 0 - -[erf] -rupture_mesh_spacing = 2.0 -width_of_mfd_bin = 0.1 -area_source_discretization = 10 -complex_fault_mesh_spacing = 10 -pointsource_distance = 200.0 - -[calculation] -source_model_logic_tree_file = ../../../../../../../HazardMosaic/test_models/cea24/in/ssmLT_little_test.xml -gsim_logic_tree_file = ../gmmLT.xml -investigation_time = 1 -intensity_measure_types_and_levels = {"PGA": logscale(0.005, 3.00, 25), - "SA(0.2)": logscale(0.005, 9.00, 25), - "SA(1.0)": logscale(0.005, 3.00, 25)} -truncation_level = 3.0 -disagg_by_src = true -maximum_distance = {'default': 200.} -horiz_comp_to_geom_mean = true -use_rates = true -minimum_magnitude = 6.6 - -[disaggregation] -distance_bin_width = 20 -num_epsilon_bins = 10 -mag_bin_width = 0.1 -epsilon_star = true -postproc_func = compute_rtgm.main -asce_version = ASCE7-22 - -[output] -hazard_maps = true -uniform_hazard_spectra = true -poes = 0.000404 0.001025 0.002105 0.004453 0.013767 diff --git a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/sites_CEA_dos.csv b/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/sites_CEA_dos.csv deleted file mode 100644 index 62c36eecdb8e..000000000000 --- a/openquake/qa_tests_data/mosaic/Projects/AELO/aeloy3/py/Run_Jobs/CEA/site10/sites_CEA_dos.csv +++ /dev/null @@ -1,2 +0,0 @@ -custom_site_id,lon,lat,vs30,z1pt0,z2pt5,vs30measured -10,76.953324,43.234055,800,31.070148646647844,0.5722405600686729,1 diff --git a/openquake/qa_tests_data/mosaic/workflow.py b/openquake/qa_tests_data/mosaic/workflow.py index 90ede39ba899..fa8b6d9f487e 100644 --- a/openquake/qa_tests_data/mosaic/workflow.py +++ b/openquake/qa_tests_data/mosaic/workflow.py @@ -72,11 +72,14 @@ def aelo(mosaic_dir): return save(mosaic_dir, 'AELO.toml', '\n'.join(lst)) -def ghm(mosaic_dir, legacy=False): +def ghm(mosaic_dir, models, legacy=False): "Build GHM.toml" lst = ['[workflow]\ndescription="GHM"'] - add_checkout(lst, MODELS) - for mod in MODELS: + if models == ['ALL']: + models = MODELS + + add_checkout(lst, models) + for mod in models: if legacy: # to support unzipped_mosaic_run if mod == 'CND': mod = 'CAN' From 7907ae9add4c0a50ef302fd1fc948d5e46440bc1 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 04:21:52 +0200 Subject: [PATCH 02/15] More work --- openquake/engine/aelo.py | 3 ++ openquake/engine/global_hazard.py | 15 ++++---- openquake/qa_tests_data/mosaic/workflow.py | 41 +++++++++++++--------- 3 files changed, 36 insertions(+), 23 deletions(-) diff --git a/openquake/engine/aelo.py b/openquake/engine/aelo.py index 491e0e6b10da..5bf704b8150f 100644 --- a/openquake/engine/aelo.py +++ b/openquake/engine/aelo.py @@ -83,6 +83,9 @@ def get_params_from(inputs, mosaic_dir, exclude=(), ini=None): params['mosaic_model'] = models[0] # NB: or the description is passed explicitly or it is generated from # the siteid, which must be a valid custom_site_id + if 'siteid' not in inputs: + inputs['siteid'] = 'SITE' + inputs['sites'].replace(' ', '_') + inputs['vs30'] = str(inputs['vs30']) if 'description' in inputs: params['description'] = inputs['description'] params['siteid'] = readinput.get_custom_site_id(inputs['description']) diff --git a/openquake/engine/global_hazard.py b/openquake/engine/global_hazard.py index 09dd3c5953dc..31016de23602 100755 --- a/openquake/engine/global_hazard.py +++ b/openquake/engine/global_hazard.py @@ -23,23 +23,26 @@ from openquake.qa_tests_data.mosaic import workflow from openquake.engine import engine -def main(mosaic_dir, models='ALL', toml:bool=False, cache:str='false'): + +def main(mosaic_dir, toml:bool=False, cache:str='false', kfilter:str=''): """ - Storing global SES + Storing global hazard mosaic """ - ghm_toml = workflow.ghm(mosaic_dir, models.split(','), toml) + ghm_toml = workflow.ghm(mosaic_dir) if toml: print(ghm_toml) return with patch.dict(config.directory, {'mosaic_dir': mosaic_dir}): - calc_id = engine.run_workflow(ghm_toml, {'cache': cache}) + calc_id = engine.run_workflow( + ghm_toml, {'cache': cache, 'kfilter': kfilter}) os.remove(ghm_toml) return calc_id main.mosaic_dir = 'Directory containing the hazard mosaic' -main.models = 'Models to consider (comma-separated)' -main.cache = 'Use the cache to avoid repeating correct calculations' main.toml = 'Just print the TOML code' +main.cache = 'Use the cache to avoid repeating correct calculations' +main.kfilter = 'Run only a subset of the calculations' + if __name__ == '__main__': sap.run(main) diff --git a/openquake/qa_tests_data/mosaic/workflow.py b/openquake/qa_tests_data/mosaic/workflow.py index fa8b6d9f487e..f28a8597e01b 100644 --- a/openquake/qa_tests_data/mosaic/workflow.py +++ b/openquake/qa_tests_data/mosaic/workflow.py @@ -2,7 +2,8 @@ import pandas from openquake.baselib import sap from openquake.hazardlib.geo.utils import geolocate -from openquake.hazardlib.countries import MODELS, REGIONS, country2code +from openquake.hazardlib.countries import ( + MODELS, ALIASES, REGIONS, country2code) from openquake.commonlib.readinput import read_mosaic_df @@ -72,28 +73,34 @@ def aelo(mosaic_dir): return save(mosaic_dir, 'AELO.toml', '\n'.join(lst)) -def ghm(mosaic_dir, models, legacy=False): +def extract(basedir, job_ini): + """ + Extract job.ini files from a base directory, recursively + """ + out = [] + for cwd, dirs, files in os.walk(basedir): + for model in MODELS: + if model in ALIASES: + mod = ALIASES[model] + else: + mod = model + for dirname in dirs: + if dirname == mod: + out.append((model, os.path.join(cwd, mod, job_ini))) + return out + + +def ghm(basedir): "Build GHM.toml" lst = ['[workflow]\ndescription="GHM"'] - if models == ['ALL']: - models = MODELS - - add_checkout(lst, models) - for mod in models: - if legacy: # to support unzipped_mosaic_run - if mod == 'CND': - mod = 'CAN' - elif mod in 'GLD OAT OPA': - continue - lst.append(f'[{mod}]\nini = "{mod}/in/job_vs30.ini"') - if legacy and mod == 'ARB': - # speedup slow sources - lst.append('area_source_discretization = 50') + add_checkout(lst, MODELS) + for mod, ini in extract(basedir, 'job.ini'): + lst.append(f'[{mod}]\nini = "{ini}"') lst.append('\n[success]') lst.append('func = "openquake.engine.postjobs.import_outputs"') lst.append('out_types = ["hcurves", "hmaps-stats"]') - return save(mosaic_dir, 'GHM.toml', '\n'.join(lst)) + return save(basedir, 'GHM.toml', '\n'.join(lst)) def grm(mosaic_dir, number_of_logic_tree_samples: int = 2000, From af597d3f456822a5a661c152d6143b1103edb15c Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 04:35:37 +0200 Subject: [PATCH 03/15] More work --- openquake/qa_tests_data/mosaic/workflow.py | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/openquake/qa_tests_data/mosaic/workflow.py b/openquake/qa_tests_data/mosaic/workflow.py index f28a8597e01b..2d1915657720 100644 --- a/openquake/qa_tests_data/mosaic/workflow.py +++ b/openquake/qa_tests_data/mosaic/workflow.py @@ -6,6 +6,8 @@ MODELS, ALIASES, REGIONS, country2code) from openquake.commonlib.readinput import read_mosaic_df +MODELDIRS = [ALIASES.get(model, model) for model in MODELS] # JPN->JPA + TOML = '''\ [workflow] @@ -33,9 +35,9 @@ def save(mosaic_dir, name, toml): return fname -def add_checkout(lst, models): - for mod in models: - lst.append(f'checkout.{mod} = "master"') +def add_checkout(lst, repos, branch="v2026_updates"): + for repo in repos: + lst.append(f'checkout.{repo} = "{branch}"') lst.append('') @@ -61,7 +63,7 @@ def aelo(mosaic_dir): sites, siteid = get_aelo_sites(site_file) lst = ['[workflow]\ndescription="AELO"'] models = [mod for mod in MODELS if mod not in {'USA', 'ALS', 'HAW'}] - add_checkout(lst, models) + add_checkout(lst, MODELDIRS) for mod in models: if mod == 'CND': mod = 'CAN' @@ -79,15 +81,11 @@ def extract(basedir, job_ini): """ out = [] for cwd, dirs, files in os.walk(basedir): - for model in MODELS: - if model in ALIASES: - mod = ALIASES[model] - else: - mod = model + for model, mod in zip(MODELS, MODELDIRS): for dirname in dirs: if dirname == mod: out.append((model, os.path.join(cwd, mod, job_ini))) - return out + return sorted(out) def ghm(basedir): From 5130d7acbd393be2f480886e599fe593a5512cb1 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 04:39:47 +0200 Subject: [PATCH 04/15] Fix --- openquake/qa_tests_data/mosaic/workflow.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/openquake/qa_tests_data/mosaic/workflow.py b/openquake/qa_tests_data/mosaic/workflow.py index 2d1915657720..d969822f254c 100644 --- a/openquake/qa_tests_data/mosaic/workflow.py +++ b/openquake/qa_tests_data/mosaic/workflow.py @@ -81,8 +81,10 @@ def extract(basedir, job_ini): """ out = [] for cwd, dirs, files in os.walk(basedir): + if '.git' in dirs: + dirs.remove('.git') for model, mod in zip(MODELS, MODELDIRS): - for dirname in dirs: + for dirname in dirs: if dirname == mod: out.append((model, os.path.join(cwd, mod, job_ini))) return sorted(out) @@ -91,7 +93,7 @@ def extract(basedir, job_ini): def ghm(basedir): "Build GHM.toml" lst = ['[workflow]\ndescription="GHM"'] - add_checkout(lst, MODELS) + add_checkout(lst, MODELDIRS) for mod, ini in extract(basedir, 'job.ini'): lst.append(f'[{mod}]\nini = "{ini}"') From fe01fc62a9b00cd2a84eed324f4ffee4e35c6d87 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 04:42:21 +0200 Subject: [PATCH 05/15] Fixed argparsing --- openquake/engine/global_hazard.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openquake/engine/global_hazard.py b/openquake/engine/global_hazard.py index 31016de23602..eb2b71c3aedf 100755 --- a/openquake/engine/global_hazard.py +++ b/openquake/engine/global_hazard.py @@ -24,7 +24,7 @@ from openquake.engine import engine -def main(mosaic_dir, toml:bool=False, cache:str='false', kfilter:str=''): +def main(mosaic_dir, toml:bool=False, *, cache:str='false', kfilter:str=''): """ Storing global hazard mosaic """ From eef7273ba7b9841eeb45ec7a1b6766c4c66d5ebb Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 04:48:20 +0200 Subject: [PATCH 06/15] Skipped padding directory --- openquake/qa_tests_data/mosaic/workflow.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/openquake/qa_tests_data/mosaic/workflow.py b/openquake/qa_tests_data/mosaic/workflow.py index d969822f254c..404bfea3b0fd 100644 --- a/openquake/qa_tests_data/mosaic/workflow.py +++ b/openquake/qa_tests_data/mosaic/workflow.py @@ -81,8 +81,9 @@ def extract(basedir, job_ini): """ out = [] for cwd, dirs, files in os.walk(basedir): - if '.git' in dirs: - dirs.remove('.git') + for skipdir in [".git", "padding"]: + if skipdir in dirs: + dirs.remove(skipdir) for model, mod in zip(MODELS, MODELDIRS): for dirname in dirs: if dirname == mod: From 727d1b93bed928a228f734dd8a01c1e107324756 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 04:50:10 +0200 Subject: [PATCH 07/15] Fix --- openquake/qa_tests_data/mosaic/workflow.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openquake/qa_tests_data/mosaic/workflow.py b/openquake/qa_tests_data/mosaic/workflow.py index 404bfea3b0fd..094903de8038 100644 --- a/openquake/qa_tests_data/mosaic/workflow.py +++ b/openquake/qa_tests_data/mosaic/workflow.py @@ -87,7 +87,7 @@ def extract(basedir, job_ini): for model, mod in zip(MODELS, MODELDIRS): for dirname in dirs: if dirname == mod: - out.append((model, os.path.join(cwd, mod, job_ini))) + out.append((model, os.path.join(cwd, mod, 'in', job_ini))) return sorted(out) From 1b2bb6e587a8fe121342bb7488350040af0fcd45 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 04:53:35 +0200 Subject: [PATCH 08/15] Sampling sites --- openquake/engine/global_hazard.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/openquake/engine/global_hazard.py b/openquake/engine/global_hazard.py index eb2b71c3aedf..71b4eede2df6 100755 --- a/openquake/engine/global_hazard.py +++ b/openquake/engine/global_hazard.py @@ -32,7 +32,8 @@ def main(mosaic_dir, toml:bool=False, *, cache:str='false', kfilter:str=''): if toml: print(ghm_toml) return - with patch.dict(config.directory, {'mosaic_dir': mosaic_dir}): + with (patch.dict(config.directory, {'mosaic_dir': mosaic_dir}), + patch.dict(os.environ, OQ_SAMPLE_SITES=".001")): calc_id = engine.run_workflow( ghm_toml, {'cache': cache, 'kfilter': kfilter}) os.remove(ghm_toml) From 98b0cf7a0af7620b2710d1b963b542a54af6a054 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 04:55:18 +0200 Subject: [PATCH 09/15] Cleanup --- openquake/engine/global_hazard.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openquake/engine/global_hazard.py b/openquake/engine/global_hazard.py index 71b4eede2df6..4cf5b3fdc080 100755 --- a/openquake/engine/global_hazard.py +++ b/openquake/engine/global_hazard.py @@ -32,7 +32,7 @@ def main(mosaic_dir, toml:bool=False, *, cache:str='false', kfilter:str=''): if toml: print(ghm_toml) return - with (patch.dict(config.directory, {'mosaic_dir': mosaic_dir}), + with (patch.dict(config.directory, mosaic_dir=mosaic_dir), patch.dict(os.environ, OQ_SAMPLE_SITES=".001")): calc_id = engine.run_workflow( ghm_toml, {'cache': cache, 'kfilter': kfilter}) From b98045fa8f444160f32fab42b5d362a126e3fd1b Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 06:48:20 +0200 Subject: [PATCH 10/15] Only importing the hmaps --- openquake/engine/postjobs.py | 5 ++++- openquake/qa_tests_data/mosaic/workflow.py | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/openquake/engine/postjobs.py b/openquake/engine/postjobs.py index 6f394ba66c36..3b79adc23142 100644 --- a/openquake/engine/postjobs.py +++ b/openquake/engine/postjobs.py @@ -151,7 +151,10 @@ def save_performance(dstore, calcs, operations): def main(postjob: str, workflow_id: int, calc_id: int, arg: str): """ - Useful for debugging errors in postjobs + Useful for debugging errors in postjobs. Here is an example for + hazard curves in mosaic calculations: + + $ oq shell openquake.engine.postjobs import_outputs 319 330 '["hcurves"]' """ postjob = globals()[postjob] with datastore.read(workflow_id, 'r+') as dstore: diff --git a/openquake/qa_tests_data/mosaic/workflow.py b/openquake/qa_tests_data/mosaic/workflow.py index 094903de8038..b6b2a92fefb9 100644 --- a/openquake/qa_tests_data/mosaic/workflow.py +++ b/openquake/qa_tests_data/mosaic/workflow.py @@ -100,7 +100,7 @@ def ghm(basedir): lst.append('\n[success]') lst.append('func = "openquake.engine.postjobs.import_outputs"') - lst.append('out_types = ["hcurves", "hmaps-stats"]') + lst.append('out_types = ["hmaps-stats"]') return save(basedir, 'GHM.toml', '\n'.join(lst)) From a3c3cf88675676522d3a9ab744fd3484c247a994 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 09:02:22 +0200 Subject: [PATCH 11/15] Extended workflow.ghm to work with the global_risk_model --- openquake/qa_tests_data/mosaic/workflow.py | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/openquake/qa_tests_data/mosaic/workflow.py b/openquake/qa_tests_data/mosaic/workflow.py index b6b2a92fefb9..1fb7273dc09e 100644 --- a/openquake/qa_tests_data/mosaic/workflow.py +++ b/openquake/qa_tests_data/mosaic/workflow.py @@ -37,7 +37,7 @@ def save(mosaic_dir, name, toml): def add_checkout(lst, repos, branch="v2026_updates"): for repo in repos: - lst.append(f'checkout.{repo} = "{branch}"') + lst.append(f'checkout."{repo}" = "{branch}"') lst.append('') @@ -79,23 +79,24 @@ def extract(basedir, job_ini): """ Extract job.ini files from a base directory, recursively """ - out = [] + out = {} for cwd, dirs, files in os.walk(basedir): - for skipdir in [".git", "padding"]: + for skipdir in [".git", "padding", "scenarios"]: if skipdir in dirs: dirs.remove(skipdir) for model, mod in zip(MODELS, MODELDIRS): for dirname in dirs: if dirname == mod: - out.append((model, os.path.join(cwd, mod, 'in', job_ini))) - return sorted(out) + out[model] = os.path.join(cwd, mod, 'in', job_ini) + return sorted(out.items()) def ghm(basedir): "Build GHM.toml" lst = ['[workflow]\ndescription="GHM"'] - add_checkout(lst, MODELDIRS) - for mod, ini in extract(basedir, 'job.ini'): + mod_inis = extract(basedir, 'job.ini') + add_checkout(lst, [os.path.dirname(ini) for _mod, ini in mod_inis]) + for mod, ini in mod_inis: lst.append(f'[{mod}]\nini = "{ini}"') lst.append('\n[success]') From 4a89b3019b8961690c8a837e4d2d8bf729bbbe34 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 10:06:11 +0200 Subject: [PATCH 12/15] Cleanup [ci skip] --- openquake/qa_tests_data/mosaic/workflow.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/openquake/qa_tests_data/mosaic/workflow.py b/openquake/qa_tests_data/mosaic/workflow.py index b6b2a92fefb9..c818f97089f3 100644 --- a/openquake/qa_tests_data/mosaic/workflow.py +++ b/openquake/qa_tests_data/mosaic/workflow.py @@ -92,7 +92,7 @@ def extract(basedir, job_ini): def ghm(basedir): - "Build GHM.toml" + "Build generatedGHM.toml" lst = ['[workflow]\ndescription="GHM"'] add_checkout(lst, MODELDIRS) for mod, ini in extract(basedir, 'job.ini'): @@ -101,7 +101,7 @@ def ghm(basedir): lst.append('\n[success]') lst.append('func = "openquake.engine.postjobs.import_outputs"') lst.append('out_types = ["hmaps-stats"]') - return save(basedir, 'GHM.toml', '\n'.join(lst)) + return save(basedir, 'generatedGHM.toml', '\n'.join(lst)) def grm(mosaic_dir, number_of_logic_tree_samples: int = 2000, From a656664e5fb3d277ebe9498d4b61c683360711d7 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 11:02:56 +0200 Subject: [PATCH 13/15] Fixed multispawn --- openquake/baselib/parallel.py | 24 ++++------------------ openquake/calculators/base.py | 6 ++++-- openquake/engine/engine.py | 11 +++++----- openquake/engine/global_hazard.py | 11 +++++----- openquake/qa_tests_data/mosaic/workflow.py | 4 ++-- 5 files changed, 21 insertions(+), 35 deletions(-) diff --git a/openquake/baselib/parallel.py b/openquake/baselib/parallel.py index 9c91aa5cf6a1..958e82a75ac8 100644 --- a/openquake/baselib/parallel.py +++ b/openquake/baselib/parallel.py @@ -198,7 +198,6 @@ def count(word): import collections from unittest import mock import multiprocessing.dummy -from multiprocessing.connection import wait import multiprocessing.shared_memory as shmem import psutil import numpy @@ -1102,29 +1101,14 @@ def multispawn(func, allargs, nprocs=num_cores, logfinish=True, :param logfinish: if True, log a progress message :param names: optionally, give names to the spawned processes """ - if names: - assert len(names) == len(allargs), (len(names), len(allargs)) if oq_distribute() == 'no': for args in allargs: func(*args) return - tot = len(allargs) - procs = {} # sentinel -> process - n = 1 - while allargs: - args = allargs.pop(0) - name = names.pop(0) if names else None - proc = mp_context.Process(target=func, args=args, name=name) - proc.start() - procs[proc.sentinel] = proc - while procs: - for finished in wait(procs): - name = procs[finished].name or '' - procs[finished].join() - del procs[finished] - if logfinish: - logging.info('Finished job %s [%d of %d]', name, n, tot) - n += 1 + + with mp_context.Pool(nprocs) as pool: + for res in pool.imap_unordered(func, allargs): + logging.info('Finished %(name)s' % res) if oq_distribute() == 'slurm': diff --git a/openquake/calculators/base.py b/openquake/calculators/base.py index 62c1ee8c1684..d6d2877ee583 100644 --- a/openquake/calculators/base.py +++ b/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): """ diff --git a/openquake/engine/engine.py b/openquake/engine/engine.py index 2f8abbb627d3..22dcb24246d5 100644 --- a/openquake/engine/engine.py +++ b/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) diff --git a/openquake/engine/global_hazard.py b/openquake/engine/global_hazard.py index 4cf5b3fdc080..63ed7e0cc854 100755 --- a/openquake/engine/global_hazard.py +++ b/openquake/engine/global_hazard.py @@ -19,27 +19,28 @@ import os from unittest.mock import patch -from openquake.baselib import sap, config +from openquake.baselib import sap from openquake.qa_tests_data.mosaic import workflow from openquake.engine import engine -def main(mosaic_dir, toml:bool=False, *, cache:str='false', kfilter:str=''): +def main(mosaic_dir, ini='job.ini', toml:bool=False, *, + cache:str='false', kfilter:str=''): """ Storing global hazard mosaic """ - ghm_toml = workflow.ghm(mosaic_dir) + ghm_toml = workflow.ghm(mosaic_dir, ini) if toml: print(ghm_toml) return - with (patch.dict(config.directory, mosaic_dir=mosaic_dir), - patch.dict(os.environ, OQ_SAMPLE_SITES=".001")): + with patch.dict(os.environ, OQ_SAMPLE_SITES=".001"): calc_id = engine.run_workflow( ghm_toml, {'cache': cache, 'kfilter': kfilter}) os.remove(ghm_toml) return calc_id main.mosaic_dir = 'Directory containing the hazard mosaic' +main.ini = 'Consider the .ini files' main.toml = 'Just print the TOML code' main.cache = 'Use the cache to avoid repeating correct calculations' main.kfilter = 'Run only a subset of the calculations' diff --git a/openquake/qa_tests_data/mosaic/workflow.py b/openquake/qa_tests_data/mosaic/workflow.py index 8cf5a8ff7a10..a72365e48356 100644 --- a/openquake/qa_tests_data/mosaic/workflow.py +++ b/openquake/qa_tests_data/mosaic/workflow.py @@ -91,10 +91,10 @@ def extract(basedir, job_ini): return sorted(out.items()) -def ghm(basedir): +def ghm(basedir, job_ini='job.ini'): "Build generatedGHM.toml" lst = ['[workflow]\ndescription="GHM"'] - mod_inis = extract(basedir, 'job.ini') + mod_inis = extract(basedir, job_ini) add_checkout(lst, [os.path.dirname(ini) for _mod, ini in mod_inis]) for mod, ini in mod_inis: lst.append(f'[{mod}]\nini = "{ini}"') From ac0abd85dde508c997ed9699f123ffeeabe13f50 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Tue, 16 Jun 2026 11:04:53 +0200 Subject: [PATCH 14/15] Skipped test --- openquake/engine/tests/global_hazard_test.py | 33 ++++++++++++++++++++ 1 file changed, 33 insertions(+) create mode 100644 openquake/engine/tests/global_hazard_test.py diff --git a/openquake/engine/tests/global_hazard_test.py b/openquake/engine/tests/global_hazard_test.py new file mode 100644 index 000000000000..95e0c2f263fb --- /dev/null +++ b/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 . + +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() From a2c8ea02f2f36663a98fc2d902c323205ceb04a4 Mon Sep 17 00:00:00 2001 From: Michele Simionato Date: Wed, 17 Jun 2026 09:56:36 +0200 Subject: [PATCH 15/15] Restored the multispawn of engine-3.25 --- openquake/baselib/parallel.py | 33 +++++++++++++++++++++++++++++---- 1 file changed, 29 insertions(+), 4 deletions(-) diff --git a/openquake/baselib/parallel.py b/openquake/baselib/parallel.py index 958e82a75ac8..e17d4879065a 100644 --- a/openquake/baselib/parallel.py +++ b/openquake/baselib/parallel.py @@ -198,6 +198,7 @@ def count(word): import collections from unittest import mock import multiprocessing.dummy +from multiprocessing.connection import wait import multiprocessing.shared_memory as shmem import psutil import numpy @@ -1101,14 +1102,38 @@ def multispawn(func, allargs, nprocs=num_cores, logfinish=True, :param logfinish: if True, log a progress message :param names: optionally, give names to the spawned processes """ + if names: + assert len(names) == len(allargs), (len(names), len(allargs)) if oq_distribute() == 'no': for args in allargs: func(*args) return - - with mp_context.Pool(nprocs) as pool: - for res in pool.imap_unordered(func, allargs): - logging.info('Finished %(name)s' % res) + tot = len(allargs) + allargs = allargs[::-1] # so that the first argument is submitted first + procs = {} # sentinel -> process + n = 1 + while allargs: + 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 '' + procs[finished].join() + del procs[finished] + if logfinish: + logging.info('Finished job %s [%d of %d]', name, n, tot) + n += 1 if oq_distribute() == 'slurm':