diff --git a/openquake/calculators/classical.py b/openquake/calculators/classical.py
index 52fc1c785223..a7d97bc851dc 100644
--- a/openquake/calculators/classical.py
+++ b/openquake/calculators/classical.py
@@ -645,14 +645,59 @@ def _execute(self, sgs, ds):
OQ_TASK_NO = os.environ.get('OQ_TASK_NO', '')
if OQ_TASK_NO:
allargs = [allargs[int(OQ_TASK_NO)]]
- if self.few_sites or oq.disagg_by_src:
- smap = parallel.Starmap(
- classical_disagg, allargs, h5=self.datastore.hdf5)
+ task_func = (classical_disagg if (self.few_sites or oq.disagg_by_src)
+ else classical)
+ if oq.sequential_source_models and not OQ_TASK_NO:
+ acc = self._run_sequential_source_models(allargs, task_func)
else:
- smap = parallel.Starmap(classical, allargs, h5=self.datastore.hdf5)
- acc = smap.reduce(self.agg_dicts, AccumDict(accum=0.))
+ smap = parallel.Starmap(
+ task_func, allargs, h5=self.datastore.hdf5)
+ acc = smap.reduce(self.agg_dicts, AccumDict(accum=0.))
self._post_execute(acc)
+ def _run_sequential_source_models(self, allargs, task_func):
+ """
+ Run one Starmap per sourceModel branch sequentially, so that
+ only a single source model's tasks are running at one time.
+
+ NOTE: Unless extendModel is being used, the sharing of src_groups
+ over base source models is not permitted. If extendModel is being
+ used, then the sharing of src_groups over base source models is
+ permitted and a "shared" batch is dispatched last. The memory
+ footprint of this "shared" batch could be similar (or equal) to
+ that observed in the "regular" (i.e., none-sequential) approach
+ if extendModel is heavily used in the logic tree (because many
+ of the src_grps would be piled into this final "shared" batch).
+ """
+ # Map each src_group id to its sourceModel branch_id
+ smb_of_grp = {
+ gid: smb
+ for smb, gids in self.csm.grp_ids_by_source_model().items()
+ for gid in gids}
+
+ # Partition the task-arg tuples by sourceModel branch
+ partitions = AccumDict(accum=[])
+ for args in allargs:
+ # Strip any tile suffix ("5-2" -> 5) to recover grp_id
+ gid = int(args[0][0].split('-')[0])
+ partitions[smb_of_grp[gid]].append(args)
+
+ # Sort per-source model partitions for reproducible
+ # order with "shared" batch last
+ per_sm_keys = sorted(k for k in partitions if k is not None)
+ ordered_keys = per_sm_keys + (
+ [None] if None in partitions else [])
+
+ # Run one source model at a time
+ acc = AccumDict(accum=0.)
+ for smb in ordered_keys:
+ part = partitions[smb]
+ logging.info('Source model %r: %d tasks', smb, len(part))
+ smap = parallel.Starmap(task_func, part, h5=self.datastore.hdf5)
+ acc = smap.reduce(self.agg_dicts, acc)
+
+ return acc
+
def _post_execute(self, acc):
# save the rates and performs some checks
oq = self.oqparam
diff --git a/openquake/calculators/preclassical.py b/openquake/calculators/preclassical.py
index 1625262cb6c0..1a15afc84a41 100644
--- a/openquake/calculators/preclassical.py
+++ b/openquake/calculators/preclassical.py
@@ -298,22 +298,16 @@ def populate_csm(self):
self.store()
logging.info('Building cmakers')
trt_smrs = csm.get_trt_smrs()
- self.cmakers = get_cmakers(trt_smrs, csm.full_lt, oq)
self.datastore.hdf5.save_vlen('trt_smrs', trt_smrs)
+ if oq.sequential_source_models:
+ grp_ids_by_batch = [
+ numpy.array(
+ [sg.sources[0].grp_id for sg in sgs_batch], U32)
+ for _, _, sgs_batch, _ in csm.iter_source_model_batches()]
+ self.datastore.hdf5.save_vlen('grp_ids_by_batch', grp_ids_by_batch)
sites = csm.sitecol if csm.sitecol else None
if sites is None:
logging.warning('No sites??')
-
- L = oq.imtls.size
- Gfull = self.full_lt.gfull([cm.trt_smrs for cm in self.cmakers])
- Gt = sum(len(cm.gsims) for cm in self.cmakers)
- extra = f'<{Gfull}' if Gt < Gfull else ''
- if sites is not None:
- nbytes = 4 * len(self.sitecol) * L * Gt
- # Gt is known before starting the preclassical
- logging.warning(f'Global RateMap of %s ({Gt=}%s)',
- general.humansize(nbytes), extra)
-
if sites and not self.few_sites:
# in SAM from 539,831 -> 11,430 sites
lowres = sites.lower_res(res=4)[0] # res=4 ~39 km
@@ -324,43 +318,99 @@ def populate_csm(self):
sf = SourceFilter(sites, oq.maximum_distance)
else:
sf = SourceFilter(None)
- atomic_sources = []
- normal_sources = []
reqv = 'reqv' in oq.inputs
if reqv:
logging.warning(
'Using equivalent distance approximation and '
'collapsing hypocenters and nodal planes')
- multifaults = []
+ multifaults = [src for sg in csm.src_groups for src in sg
+ if src.code == b'F']
+ if multifaults:
+ with hdf5.File(multifaults[0].hdf5path, 'r') as h5:
+ secparams = h5['secparams'][:]
+ logging.warning(
+ 'There are %d multiFaultSources (secparams=%s)',
+ len(multifaults), general.humansize(secparams.nbytes))
+ else:
+ secparams = ()
+ if oq.sequential_source_models:
+ # Bound preclassical memory by iterating one source model
+ # at a time and rebuild cmakers at end
+ self._run_batched(sf, secparams, reqv)
+ self.cmakers = get_cmakers(trt_smrs, csm.full_lt, oq)
+ else:
+ self._run_regular(trt_smrs, sf, secparams, reqv)
+ L = oq.imtls.size
+ Gfull = self.full_lt.gfull([cm.trt_smrs for cm in self.cmakers])
+ Gt = sum(len(cm.gsims) for cm in self.cmakers)
+ extra = f'<{Gfull}' if Gt < Gfull else ''
+ if sites is not None:
+ nbytes = 4 * len(self.sitecol) * L * Gt
+ logging.warning(f'Global RateMap of %s ({Gt=}%s)',
+ general.humansize(nbytes), extra)
+ allsources = csm.get_sources()
+ self.store_source_info(source_data(allsources))
+
+ def _run_batched(self, sf, secparams, reqv):
+ """
+ Run preclassical per source-model batch when the flag of
+ sequential_source_models is True.
+ """
+ oq = self.oqparam
+ csm = self.csm
+ for batch_id, sm_id, sgs_batch, trt_smrs_batch in (
+ csm.iter_source_model_batches()):
+ logging.info(
+ 'Preclassical batch %d (source model %r): %d src_groups',
+ batch_id, sm_id, len(sgs_batch))
+ cmakers_batch = get_cmakers(trt_smrs_batch, csm.full_lt, oq)
+ cmaker_by_grp = {
+ sg.sources[0].grp_id: cm
+ for sg, cm in zip(sgs_batch, cmakers_batch.to_array())}
+ atomic_batch = []
+ normal_batch = []
+ for sg in sgs_batch:
+ for src in sg:
+ if reqv and sg.trt in oq.inputs['reqv']:
+ if src.source_id not in oq.reqv_ignore_sources:
+ collapse_nphc(src)
+ grp_id = sg.sources[0].grp_id
+ if sg.atomic:
+ cmaker_by_grp[grp_id].set_weight(sg, sf)
+ atomic_batch.extend(sg)
+ else:
+ normal_batch.extend(sg)
+ self._process(atomic_batch, normal_batch, sf, secparams,
+ cmaker_by_grp=cmaker_by_grp)
+
+ def _run_regular(self, trt_smrs, sf, secparams, reqv):
+ """
+ Run preclassical in a single pass over all src_groups when
+ sequential_source_models is False.
+ """
+ oq = self.oqparam
+ csm = self.csm
+ self.cmakers = get_cmakers(trt_smrs, csm.full_lt, oq)
+ atomic_sources = []
+ normal_sources = []
cmakers = self.cmakers.to_array()
for sg in csm.src_groups:
for src in sg:
- if src.code == b'F':
- multifaults.append(src)
if reqv and sg.trt in oq.inputs['reqv']:
if src.source_id not in oq.reqv_ignore_sources:
collapse_nphc(src)
grp_id = sg.sources[0].grp_id
- # do nothing for atomic sources except counting the ruptures
if sg.atomic:
- # compute weight sequentially
cmakers[grp_id].set_weight(sg, sf)
atomic_sources.extend(sg)
else:
normal_sources.extend(sg)
- if multifaults:
- with hdf5.File(multifaults[0].hdf5path, 'r') as h5:
- secparams = h5['secparams'][:]
- logging.warning(
- 'There are %d multiFaultSources (secparams=%s)',
- len(multifaults), general.humansize(secparams.nbytes))
- else:
- secparams = ()
self._process(atomic_sources, normal_sources, sf, secparams)
- allsources = csm.get_sources()
- self.store_source_info(source_data(allsources))
- def _process(self, atomic_sources, normal_sources, sf, secparams):
+ def _process(self, atomic_sources, normal_sources, sf, secparams,
+ cmaker_by_grp=None):
+ if cmaker_by_grp is None:
+ cmaker_by_grp = dict(enumerate(self.cmakers.to_array()))
# run preclassical in parallel for non-atomic sources
if normal_sources:
sources_by_key = groupby(
@@ -371,10 +421,9 @@ def _process(self, atomic_sources, normal_sources, sf, secparams):
# avoid a segfault in macOS
self.datastore.swmr_on()
smap = parallel.Starmap(preclassical, h5=self.datastore.hdf5)
- cmakers = self.cmakers.to_array()
num_tasks = len(sources_by_key)
for grp_id, srcs in sources_by_key.items():
- cmaker = cmakers[grp_id]
+ cmaker = cmaker_by_grp[grp_id]
cmaker.gsims = list(cmaker.gsims) # reducing data transfer
pointlike = [src for src in srcs
if hasattr(src, 'nodal_plane_distribution')]
diff --git a/openquake/calculators/tests/logictree_test.py b/openquake/calculators/tests/logictree_test.py
index 2d6d7df9a679..62cd07d333fe 100644
--- a/openquake/calculators/tests/logictree_test.py
+++ b/openquake/calculators/tests/logictree_test.py
@@ -32,10 +32,10 @@
from openquake.qa_tests_data.logictree import (
case_01, case_02, case_03, case_04, case_05, case_06, case_07, case_08,
case_09, case_10, case_11, case_12, case_13, case_14, case_15, case_16,
- case_17, case_18, case_19, case_20, case_21, case_22, case_23, case_25,
- case_26, case_28, case_29, case_30, case_31, case_32, case_33, case_36,
- case_39, case_45, case_46, case_52, case_56, case_58, case_59, case_67,
- case_68, case_71, case_73, case_79, case_80, case_83, case_84)
+ case_17, case_18, case_19, case_20, case_21, case_22, case_23, case_24,
+ case_25, case_26, case_28, case_29, case_30, case_31, case_32, case_33,
+ case_36, case_39, case_45, case_46, case_52, case_56, case_58, case_59,
+ case_67, case_68, case_71, case_73, case_79, case_80, case_83, case_84)
ae = numpy.testing.assert_equal
aac = numpy.testing.assert_allclose
@@ -501,6 +501,31 @@ def test_case_23_bis(self):
ns = len(self.calc.datastore['source_info'])
assert ns == 26
+ def test_case_24(self):
+ # Parity check: with sequential_source_models=true the hazard
+ # statistics (mean and quantiles) must match the regular
+ # (all-in-one Starmap) approach for both full enumeration and
+ # sampling. A small tolerance is used because task reduction
+ # order can differ across runs
+
+ # Full enumeration
+ self.run_calc(case_24.__file__, 'job.ini',
+ sequential_source_models='true')
+ seq_full = self.calc.datastore['hcurves-stats'][:]
+ self.run_calc(case_24.__file__, 'job.ini')
+ reg_full = self.calc.datastore['hcurves-stats'][:]
+ aac(seq_full, reg_full, atol=1e-6, rtol=1e-6)
+
+ # Sampling
+ self.run_calc(case_24.__file__, 'job.ini',
+ sequential_source_models='true',
+ number_of_logic_tree_samples='10')
+ seq_sampled = self.calc.datastore['hcurves-stats'][:]
+ self.run_calc(case_24.__file__, 'job.ini',
+ number_of_logic_tree_samples='10')
+ reg_sampled = self.calc.datastore['hcurves-stats'][:]
+ aac(seq_sampled, reg_sampled, atol=1e-6, rtol=1e-6)
+
def test_case_25(self):
# BCHydro-style correlated uncertainties (alt1 + alt2 + alt3)
# sampled to keep the calc fast (highly simplified version)
@@ -804,6 +829,27 @@ def test_case_83(self):
self.run_calc(case_83.__file__, 'job_expanded_LT.ini')
[fname_ex] = export(('hcurves/mean', 'csv'), self.calc.datastore)
self.assertEqualFiles(fname_em, fname_ex)
+ reg_full = self.calc.datastore['hcurves-stats'][:]
+
+ # Check that sequential approach matches regular with
+ # full enumeration
+ self.run_calc(case_83.__file__, 'job_extendModel.ini',
+ sequential_source_models='true')
+ seq_full = self.calc.datastore['hcurves-stats'][:]
+ aac(seq_full, reg_full, atol=1e-6, rtol=1e-6)
+
+ # Run regular approach with sampling
+ self.run_calc(case_83.__file__, 'job_extendModel.ini',
+ number_of_logic_tree_samples='10')
+ reg_sampled = self.calc.datastore['hcurves-stats'][:]
+
+ # Check that sequential approach matches regular with
+ # sampling
+ self.run_calc(case_83.__file__, 'job_extendModel.ini',
+ sequential_source_models='true',
+ number_of_logic_tree_samples='10')
+ seq_sampled = self.calc.datastore['hcurves-stats'][:]
+ aac(seq_sampled, reg_sampled, atol=1e-6, rtol=1e-6)
def test_case_83_eb(self):
# event based sampling with double extendModel
diff --git a/openquake/commonlib/oqvalidation.py b/openquake/commonlib/oqvalidation.py
index 48b4139513ea..7b28891bc000 100644
--- a/openquake/commonlib/oqvalidation.py
+++ b/openquake/commonlib/oqvalidation.py
@@ -793,6 +793,14 @@
Example: *ses_seed = 123*.
Default: 42
+sequential_source_models:
+ Flag used in classical and disaggregation calculations to dispatch
+ tasks one top-level sourceModel branch at a time (one Starmap per
+ source model, run sequentially). Not compatible with source models
+ that share sources across top-level branches.
+ Example: *sequential_source_models = true*.
+ Default: false
+
shakemap_id:
Used in ShakeMap calculations to download a ShakeMap from the USGS site
Example: *shakemap_id = usp000fjta*.
@@ -1261,6 +1269,7 @@ class OqParam(valid.ParamSet):
ses_per_logic_tree_path = valid.Param(
valid.compose(valid.nonzero, valid.positiveint), 1)
ses_seed = valid.Param(valid.positiveint, 42)
+ sequential_source_models = valid.Param(valid.boolean, False)
shakemap_id = valid.Param(valid.nice_string, None)
# example: shakemap_uri = {'kind': 'usgs_id', 'id': 'XXX'}
shakemap_uri = valid.Param(valid.dictionary, {})
@@ -2315,6 +2324,15 @@ def is_valid_disagg_by_src(self):
return self.ps_grid_spacing == 0
return True
+ def is_valid_sequential_source_models(self):
+ """
+ sequential_source_models is only useable in classical and
+ disaggregation calculations
+ """
+ if self.sequential_source_models:
+ return self.calculation_mode in ('classical', 'disaggregation')
+ return True
+
def is_valid_concurrent_tasks(self):
"""
At most you can use 30_000 tasks
diff --git a/openquake/commonlib/tests/source_test.py b/openquake/commonlib/tests/source_test.py
index 05d21b41887f..2864df1ace96 100644
--- a/openquake/commonlib/tests/source_test.py
+++ b/openquake/commonlib/tests/source_test.py
@@ -17,6 +17,7 @@
# along with OpenQuake. If not, see .
import os
+import copy
import unittest
from io import BytesIO
@@ -29,7 +30,10 @@
site, geo, mfd, pmf, scalerel, valid, tests as htests)
from openquake.hazardlib import source, sourceconverter as s
from openquake.hazardlib.tom import PoissonTOM
-from openquake.hazardlib.logictree import FullLogicTree
+from openquake.hazardlib.lt import Realization, BranchSet
+from openquake.hazardlib.logictree import FullLogicTree, SourceModelLogicTree
+from openquake.hazardlib.source_group import CompositeSourceModel, SourceGroup
+from openquake.hazardlib.source_reader import sampling_dt
from openquake.hazardlib import nrml
from openquake.commonlib import tests, readinput
@@ -751,3 +755,118 @@ def test_oversampling(self):
def tearDown(self):
Starmap.shutdown()
+
+
+class SequentialSourcesTestCase(unittest.TestCase):
+ """
+ Tests for the sequential_source_models dispatch including the
+ grp_ids_by_source_model method, which is used to partition
+ src_groups by top-level sourceModel branch.
+ """
+ @classmethod
+ def setUpClass(cls):
+ # Load a real source, and once per-test have
+ # their sampling param set to control trt_smrs
+ conv = s.SourceConverter(investigation_time=50.,
+ rupture_mesh_spacing=1,
+ complex_fault_mesh_spacing=1,
+ width_of_mfd_bin=1.,
+ area_source_discretization=1.)
+ [point_grp, *_] = nrml.to_python(MIXED_SRC_MODEL, conv)
+ cls.template_src = point_grp[0]
+
+ def _build_csm(self, sm_branch_ids, smrs_per_group, has_extend=False):
+ # sm_branch_ids[i] = branch id of smr i; smrs_per_group[j] =
+ # smrs of the j-th src_group
+
+ # One Realization per smr
+ sm_rlzs = [Realization(bid, 1., i, (bid,))
+ for i, bid in enumerate(sm_branch_ids)]
+ full_lt = object.__new__(FullLogicTree)
+ full_lt.sm_rlzs = sm_rlzs
+
+ # Minimal source_model_lt
+ utypes = ['sourceModel'] + (['extendModel'] if has_extend else [])
+ smlt = object.__new__(SourceModelLogicTree)
+ smlt.branchsets = [BranchSet(ut) for ut in utypes]
+ full_lt.source_model_lt = smlt
+
+ # One SourceGroup per entry
+ src_groups = []
+ for smrs in smrs_per_group:
+ src = copy.copy(self.template_src)
+ # Pack given smrs into the sampling array
+ src.sampling = numpy.array(
+ [(smr, 1) for smr in smrs], sampling_dt)
+ # Empty SourceGroup
+ sg = SourceGroup(self.template_src.tectonic_region_type)
+ # Store sources
+ sg.sources = [src]
+ src_groups.append(sg)
+
+ # Make the CSM
+ csm = object.__new__(CompositeSourceModel)
+ csm.full_lt = full_lt
+ csm.src_groups = src_groups
+
+ return csm
+
+ def test_grp_ids_by_source_model_disjoint(self):
+ # Two src_groups, each belonging to a distinct top-level
+ # source model, group cleanly by branch_id
+ csm = self._build_csm(sm_branch_ids=['sm_a', 'sm_b'],
+ smrs_per_group=[[0], [1]])
+ self.assertEqual(csm.grp_ids_by_source_model(),
+ {'sm_a': [0], 'sm_b': [1]})
+
+ def test_grp_ids_by_source_model_shared_raises_without_extend(self):
+ # A src_group whose trt_smrs point at smrs from more than
+ # one top-level sourceModel branch must raise an error
+ # if extendModel is not used in the logic tree
+ csm = self._build_csm(sm_branch_ids=['sm_a', 'sm_b'],
+ smrs_per_group=[[0, 1]])
+ with self.assertRaises(ValueError) as cm:
+ csm.grp_ids_by_source_model()
+ self.assertEqual(
+ str(cm.exception),
+ "src_group 0 (Stable Continental Crust) spans multiple "
+ "source models ['sm_a', 'sm_b']; "
+ "sequential_source_models=true does not support sources "
+ "shared across source models outside of extendModel"
+ )
+
+ def test_grp_ids_by_source_model_shared_with_extend(self):
+ # With extendModel the src_grp sharing is permitted and
+ # it results in a cross-SM group under a key of None
+ csm = self._build_csm(sm_branch_ids=['sm_a', 'sm_b'],
+ smrs_per_group=[[0], [1], [0, 1]],
+ has_extend=True)
+ self.assertEqual(csm.grp_ids_by_source_model(),
+ {'sm_a': [0], 'sm_b': [1], None: [2]})
+
+ def test_iter_source_model_batches(self):
+ # 12 src_groups total across three SMs plus one cross-SM group:
+ # 4 groups belong to smA only
+ # 2 groups belong to smB only
+ # 5 groups belong to smC only
+ # 1 group spans all three SMs (smrs=[0, 1, 2]) -> shared
+ # Expected: four batches, per-SM ones sorted (4, 2, 5 groups),
+ # shared batch last with the single cross-SM group (12 total)
+ csm = self._build_csm(
+ sm_branch_ids=['smA', 'smB', 'smC'],
+ smrs_per_group=[[0]]*4 + [[1]]*2 + [[2]]*5 + [[0, 1, 2]],
+ has_extend=True)
+ batches = list(csm.iter_source_model_batches())
+
+ self.assertEqual(len(batches), 4)
+ self.assertEqual([b[0] for b in batches], [0, 1, 2, 3])
+ self.assertEqual([b[1] for b in batches],
+ ['smA', 'smB', 'smC', None])
+ self.assertEqual([len(b[2]) for b in batches], [4, 2, 5, 1])
+ # Shared batch trt_smrs cover every SM
+ self.assertEqual(sorted(batches[-1][3][0].tolist()), [0, 1, 2])
+ # Batches together cover every src_group exactly once
+ union = []
+ for _, _, sgs, _ in batches:
+ union.extend(sgs)
+ self.assertEqual(set(union), set(csm.src_groups))
diff --git a/openquake/hazardlib/source_group.py b/openquake/hazardlib/source_group.py
index 8f203e4ecb55..bfeef7e81c4a 100644
--- a/openquake/hazardlib/source_group.py
+++ b/openquake/hazardlib/source_group.py
@@ -359,6 +359,47 @@ def get_sources(self, smr=None):
srcs.extend(grp)
return srcs
+ def grp_ids_by_source_model(self):
+ """
+ :returns:
+ Dict grouping src_groups by the branch_id of the top-level
+ sourceModel branch they trace back to (via trt_smrs).
+
+ NOTE: When the logic tree uses extendModel, groups whose
+ trt_smrs span more than one sourceModel branch form a "shared"
+ cross-source-model batch. Without extendModel such sharing is
+ treated as an error.
+
+ NOTE: If there is heavy usage of extendModel within a logic
+ tree, this large batch could be have a memory profile close
+ to (if not equal to) the "regular" (i.e., none sequential)
+ approach.
+ """
+ has_extend = any(
+ bset.uncertainty_type == 'extendModel'
+ for bset in self.full_lt.source_model_lt.branchsets)
+
+ # Build an smr -> top-level sourceModel branch_id map
+ smr_smb = {smr: rlz.lt_path[0]
+ for smr, rlz in enumerate(self.full_lt.sm_rlzs)}
+ out = collections.defaultdict(list)
+ for grp_id, sg in enumerate(self.src_groups):
+ smbs = {smr_smb[trt_smr % TWO24]
+ for trt_smr in sg.sources[0].trt_smrs}
+ if len(smbs) > 1:
+ if not has_extend:
+ raise ValueError(
+ 'src_group %d (%s) spans multiple source '
+ 'models %s; sequential_source_models=true '
+ 'does not support sources shared across '
+ 'source models outside of extendModel'
+ % (grp_id, sg.trt, sorted(smbs)))
+ key = None
+ else:
+ key = smbs.pop()
+ out[key].append(grp_id)
+ return out
+
def get_trt_smrs(self):
"""
:returns: an array of trt_smrs (to be stored as an hdf5.vuint32 array)
@@ -367,6 +408,35 @@ def get_trt_smrs(self):
assert len(keys) < TWO16, len(keys)
return [numpy.array(trt_smrs, numpy.uint32) for trt_smrs in keys]
+ def iter_source_model_batches(self):
+ """
+ Iterate "src_groups" in batches, one per sourceModel branch.
+
+ NOTE: A final "shared batch" (if any) is run which carries the
+ src_groups whose trt_smrs span multiple sourceModel branches
+ (i.e., when extendModel has been used - else shared src_groups
+ are forbidden in the sequential approach).
+ """
+ # Map sourceModel branch id -> list of global grp_ids
+ grp_ids_by_sm = self.grp_ids_by_source_model()
+
+ # Per-SM batches sorted for reproducible batch_id, shared last
+ per_sm_keys = sorted(k for k in grp_ids_by_sm if k is not None)
+ ordered_keys = per_sm_keys + (
+ [None] if None in grp_ids_by_sm else [])
+
+ for batch_id, sm_branch_id in enumerate(ordered_keys):
+ grp_ids = grp_ids_by_sm[sm_branch_id]
+
+ # Pick this batch's src_groups by their global grp_ids
+ src_groups_batch = [self.src_groups[gid] for gid in grp_ids]
+
+ # Per-batch trt_smrs arrays
+ trt_smrs_batch = [
+ numpy.array(sg.sources[0].trt_smrs, numpy.uint32)
+ for sg in src_groups_batch]
+ yield batch_id, sm_branch_id, src_groups_batch, trt_smrs_batch
+
def get_cmakers(self):
"""
:param oq: the OqParam used to build the CompositeSourceModel
diff --git a/openquake/qa_tests_data/logictree/README.md b/openquake/qa_tests_data/logictree/README.md
index 3457829957e8..300893d40a61 100644
--- a/openquake/qa_tests_data/logictree/README.md
+++ b/openquake/qa_tests_data/logictree/README.md
@@ -27,6 +27,7 @@
| case\_22 | Test sigma_model_alatik2015 |
| case\_23 | Arctic region and IDL (no bounding box) |
| case\_23\_bis | Correlated uncertainties |
+| case\_24 | Tests sequential_source_models parity vs regular for full-enum + sampling |
| case\_28 | Test collapse\_gsim\_logic\_tree |
| case\_28\_bis | Test missing z1pt0 |
| case\_25 | BC Hydro NVA SSC LT source model LT |
@@ -52,6 +53,6 @@
| case\_73 | Tests some epistemic uncertainties in a source-specific LT |
| case\_79 | Tests disagg\_by\_src with semicolon sources |
| case\_80 | Tests areaSourceGeometryAbsolute |
-| case\_83 | Tests extendModel and reqv |
+| case\_83 | Tests extendModel and reqv + sequential source models with extendModel |
| case\_83\_eb | Double extendModel with event based sampling |
| case\_84 | Tests maxMagGRRelativeNoMoBalance uncertainty |
diff --git a/openquake/qa_tests_data/logictree/case_24/__init__.py b/openquake/qa_tests_data/logictree/case_24/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/openquake/qa_tests_data/logictree/case_24/gsim_logic_tree.xml b/openquake/qa_tests_data/logictree/case_24/gsim_logic_tree.xml
new file mode 100644
index 000000000000..0a4537205571
--- /dev/null
+++ b/openquake/qa_tests_data/logictree/case_24/gsim_logic_tree.xml
@@ -0,0 +1,15 @@
+
+
+
+
+
+
+ BooreAtkinson2008
+ 1.0
+
+
+
+
+
diff --git a/openquake/qa_tests_data/logictree/case_24/job.ini b/openquake/qa_tests_data/logictree/case_24/job.ini
new file mode 100644
index 000000000000..1b7e7096b8ca
--- /dev/null
+++ b/openquake/qa_tests_data/logictree/case_24/job.ini
@@ -0,0 +1,39 @@
+[general]
+
+description = sequential_source_models parity test
+calculation_mode = classical
+random_seed = 42
+
+[geometry]
+
+sites = 0.0 0.0
+
+[logic_tree]
+
+number_of_logic_tree_samples = 0
+
+[erf]
+
+rupture_mesh_spacing = 2.0
+width_of_mfd_bin = 0.5
+
+[site_params]
+
+reference_vs30_type = measured
+reference_vs30_value = 760.0
+reference_depth_to_2pt5km_per_sec = 2.5
+reference_depth_to_1pt0km_per_sec = 50.0
+
+[calculation]
+
+source_model_logic_tree_file = smlt.xml
+gsim_logic_tree_file = gsim_logic_tree.xml
+investigation_time = 50.0
+intensity_measure_types_and_levels = {"PGA": [0.05, 0.1, 0.2, 0.5]}
+truncation_level = 3.0
+maximum_distance = 200.0
+
+[output]
+
+mean = true
+quantiles = 0.05 0.5 0.95
diff --git a/openquake/qa_tests_data/logictree/case_24/sm_a.xml b/openquake/qa_tests_data/logictree/case_24/sm_a.xml
new file mode 100644
index 000000000000..d5eb82e2620d
--- /dev/null
+++ b/openquake/qa_tests_data/logictree/case_24/sm_a.xml
@@ -0,0 +1,25 @@
+
+
+
+
+
+
+
+ -0.05 0.0
+ 0.05 0.0
+
+
+ 90.0
+ 0.0
+ 10.0
+
+ PeerMSR
+ 1.0
+
+ 0.0
+
+
+
diff --git a/openquake/qa_tests_data/logictree/case_24/sm_b.xml b/openquake/qa_tests_data/logictree/case_24/sm_b.xml
new file mode 100644
index 000000000000..71c68c59e546
--- /dev/null
+++ b/openquake/qa_tests_data/logictree/case_24/sm_b.xml
@@ -0,0 +1,25 @@
+
+
+
+
+
+
+
+ -0.03 0.02
+ 0.03 0.02
+
+
+ 90.0
+ 0.0
+ 10.0
+
+ PeerMSR
+ 1.0
+
+ 0.0
+
+
+
diff --git a/openquake/qa_tests_data/logictree/case_24/smlt.xml b/openquake/qa_tests_data/logictree/case_24/smlt.xml
new file mode 100644
index 000000000000..2539a4cd9001
--- /dev/null
+++ b/openquake/qa_tests_data/logictree/case_24/smlt.xml
@@ -0,0 +1,43 @@
+
+
+
+
+
+
+
+ sm_a.xml
+ 0.7
+
+
+ sm_b.xml
+ 0.3
+
+
+
+
+
+ 3.0 1.0
+ 0.6
+
+
+ 2.9 1.05
+ 0.4
+
+
+
+
+
+ 3.2 0.9
+ 0.5
+
+
+ 3.1 0.95
+ 0.5
+
+
+
+
+