Source code for openquake.commands.sample

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright (C) 2015-2023 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 shutil
import numpy
import pandas
from openquake.hazardlib import valid, nrml, sourceconverter, sourcewriter
from openquake.baselib import general
from openquake.commonlib import logictree


[docs]def save_bak(fname, node, num_nodes, total): shutil.copy(fname, fname + '.bak') print('Copied the original file in %s.bak' % fname) with open(fname, 'wb') as f: nrml.write(node, f, xmlns=node['xmlns']) print('Extracted %d nodes out of %d' % (num_nodes, total))
[docs]def reduce_source_model(fname, reduction_factor, itime): """ Reduce the source model by sampling the sources; as a special case, multiPointSources are split in pointSources and then sampled. """ conv = sourceconverter.SourceConverter(area_source_discretization=20., investigation_time=itime) [smodel] = nrml.read_source_models([fname], conv) grp = smodel.src_groups[0] if any(src.code == b'M' for src in grp): # multiPoint for src in grp: if src.code == b'M': grp.sources = general.random_filter(src, reduction_factor) print('Extracted %d point sources out of %d' % (len(grp), len(src))) break elif any(src.code == b'F' for src in grp): # multiFault for src in grp: if src.code == b'F': rids = numpy.arange(src.count_ruptures()) ok = general.random_filter(rids, reduction_factor) src.mags = src.mags[ok] src.rakes = src.rakes[ok] src.probs_occur = src.probs_occur[ok] src._rupture_idxs = [src._rupture_idxs[o] for o in ok] print('Extracted %d ruptures out of %d' % (len(ok), len(rids))) break else: total = len(grp) grp.sources = general.random_filter(grp, reduction_factor) print('Extracted %d nodes out of %d' % (len(grp), total)) smodel.src_groups = [grp] shutil.copy(fname, fname + '.bak') print('Copied the original file in %s.bak' % fname) sourcewriter.write_source_model(fname, smodel)
[docs]def main(fname, reduction_factor: valid.probability, investigation_time: valid.positivefloat = 50.): """ Produce a submodel from `fname` by sampling the nodes randomly. Supports source models, site models and exposure models. As a special case, it is also able to reduce .csv files by sampling the lines. This is a debugging utility to reduce large computations to small ones. """ if fname.endswith('.csv'): df = pandas.read_csv(fname, dtype=str) idxs = general.random_filter(numpy.arange(len(df)), reduction_factor) shutil.copy(fname, fname + '.bak') print('Copied the original file in %s.bak' % fname) df.loc[idxs].to_csv(fname, index=False, lineterminator='\r\n', na_rep='nan') print('Extracted %d lines out of %d' % (len(idxs), len(df))) return elif fname.endswith('.npy'): array = numpy.load(fname) shutil.copy(fname, fname + '.bak') print('Copied the original file in %s.bak' % fname) arr = numpy.array(general.random_filter(array, reduction_factor)) numpy.save(fname, arr) print('Extracted %d rows out of %d' % (len(arr), len(array))) return node = nrml.read(fname) model = node[0] if model.tag.endswith('exposureModel'): total = len(model.assets) model.assets.nodes = general.random_filter( model.assets, reduction_factor) num_nodes = len(model.assets) elif model.tag.endswith('siteModel'): total = len(model) model.nodes = general.random_filter(model, reduction_factor) num_nodes = len(model) elif model.tag.endswith('sourceModel'): reduce_source_model(fname, reduction_factor, investigation_time) return elif model.tag.endswith('logicTree'): for smpath in logictree.collect_info(fname).smpaths: reduce_source_model(smpath, reduction_factor, investigation_time) return else: raise RuntimeError('Unknown model tag: %s' % model.tag) save_bak(fname, node, num_nodes, total)
main.fname = 'path to the model file' main.reduction_factor = 'reduction factor in the range 0..1' main.investigation_time = 'investigation_time used in read_source_models'