Source code for openquake.commonlib.logictree

# -*- coding: utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (C) 2010-2019 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
# 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 <>.

Logic tree parser, verifier and processor. See specs at

A logic tree object must be iterable and yielding realizations, i.e. objects
with attributes `value`, `weight`, `lt_path` and `ordinal`.

import os
import re
import ast
import copy
import time
import logging
import itertools
import collections
import operator
from collections import namedtuple
import toml
import numpy
from openquake.baselib import hdf5, node, python3compat
from openquake.baselib.general import (groupby, group_array, duplicated,
import openquake.hazardlib.source as ohs
from openquake.hazardlib.gsim.base import CoeffsTable
from openquake.hazardlib.gsim.gmpe_table import GMPETable
from openquake.hazardlib.imt import from_string
from openquake.hazardlib import geo, valid, nrml, InvalidFile, pmf
from openquake.hazardlib.sourceconverter import (
    split_coords_2d, split_coords_3d, SourceGroup)

from openquake.baselib.node import node_from_elem, Node as N, context

#: Minimum value for a seed number
MIN_SINT_32 = -(2 ** 31)
#: Maximum value for a seed number
MAX_SINT_32 = (2 ** 31) - 1

TRT_REGEX = re.compile(r'tectonicRegion="([^"]+?)"')
ID_REGEX = re.compile(r'id="([^"]+?)"')
SOURCE_TYPE_REGEX = re.compile(r'<(\w+Source)\b')

[docs]def unique(objects, key=None): """ Raise a ValueError if there is a duplicated object, otherwise returns the objects as they are. """ dupl = [] for obj, group in itertools.groupby(sorted(objects), key): if sum(1 for _ in group) > 1: dupl.append(obj) if dupl: raise ValueError('Found duplicates %s' % dupl) return objects
[docs]class LtSourceModel(object): """ A container of SourceGroup instances with some additional attributes describing the source model in the logic tree. """ def __init__(self, names, weight, path, src_groups, num_gsim_paths, ordinal, samples): self.names = ' '.join(names.split()) # replace newlines with spaces self.weight = weight self.path = path self.src_groups = src_groups self.num_gsim_paths = num_gsim_paths self.ordinal = ordinal self.samples = samples @property def name(self): """ Compact representation for the names """ names = self.names.split() if len(names) == 1: return names[0] elif len(names) == 2: return ' '.join(names) else: return ' '.join([names[0], '...', names[-1]]) @property def num_sources(self): """ Number of sources contained in the source model """ return sum(len(sg) for sg in self.src_groups)
[docs] def get_skeleton(self): """ Return an empty copy of the source model, i.e. without sources, but with the proper attributes for each SourceGroup contained within. """ src_groups = [] for grp in self.src_groups: sg = copy.copy(grp) sg.sources = [] src_groups.append(sg) return self.__class__(self.names, self.weight, self.path, src_groups, self.num_gsim_paths, self.ordinal, self.samples)
def __repr__(self): samples = ', samples=%d' % self.samples if self.samples > 1 else '' return '<%s #%d %s, path=%s, weight=%s%s>' % ( self.__class__.__name__, self.ordinal, self.names, '_'.join(self.path), self.weight, samples)
Realization = namedtuple('Realization', 'value weight lt_path ordinal lt_uid') Realization.uid = property(lambda self: '_'.join(self.lt_uid)) # unique ID
[docs]def get_effective_rlzs(rlzs): """ Group together realizations with the same unique identifier (uid) and yield the first representative of each group. """ effective = [] for uid, group in groupby(rlzs, operator.attrgetter('uid')).items(): rlz = group[0] if all(path == '@' for path in rlz.lt_uid): # empty realization continue effective.append( Realization(rlz.value, sum(r.weight for r in group), rlz.lt_path, rlz.ordinal, rlz.lt_uid)) return effective
[docs]class LogicTreeError(Exception): """ Logic tree file contains a logic error. :param node: XML node object that causes fail. Used to determine the affected line number. All other constructor parameters are passed to :class:`superclass' <LogicTreeError>` constructor. """ def __init__(self, node, filename, message): self.filename = filename self.message = message self.lineno = getattr(node, 'lineno', '?') def __str__(self): return "filename '%s', line %s: %s" % ( self.filename, self.lineno, self.message)
[docs]def sample(weighted_objects, num_samples, seed): """ Take random samples of a sequence of weighted objects :param weighted_objects: A finite sequence of objects with a `.weight` attribute. The weights must sum up to 1. :param num_samples: The number of samples to return :param seed: A random seed :return: A subsequence of the original sequence with `num_samples` elements """ weights = [] for obj in weighted_objects: w = obj.weight if isinstance(obj.weight, float): weights.append(w) else: weights.append(w['weight']) numpy.random.seed(seed) idxs = numpy.random.choice(len(weights), num_samples, p=weights) # NB: returning an array would break things return [weighted_objects[idx] for idx in idxs]
[docs]class Branch(object): """ Branch object, represents a ``<logicTreeBranch />`` element. :param bs_id: BranchSetID of the branchset to which the branch belongs :param branch_id: Value of ``@branchID`` attribute. :param weight: float value of weight assigned to the branch. A text node contents of ``<uncertaintyWeight />`` child node. :param value: The actual uncertainty parameter value. A text node contents of ``<uncertaintyModel />`` child node. Type depends on the branchset's uncertainty type. """ def __init__(self, bs_id, branch_id, weight, value): self.bs_id = bs_id self.branch_id = branch_id self.weight = weight self.value = value self.child_branchset = None def __repr__(self): if self.child_branchset: return '%s%s' % (self.branch_id, self.child_branchset) else: return '%s' % self.branch_id
# Define the keywords associated with the MFD MFD_UNCERTAINTY_TYPES = ['maxMagGRRelative', 'maxMagGRAbsolute', 'bGRRelative', 'abGRAbsolute', 'incrementalMFDAbsolute'] # Define the keywords associated with the source geometry GEOMETRY_UNCERTAINTY_TYPES = ['simpleFaultDipRelative', 'simpleFaultDipAbsolute', 'simpleFaultGeometryAbsolute', 'complexFaultGeometryAbsolute', 'characteristicFaultGeometryAbsolute']
[docs]class BranchSet(object): """ Branchset object, represents a ``<logicTreeBranchSet />`` element. :param uncertainty_type: String value. According to the spec one of: gmpeModel Branches contain references to different GMPEs. Values are parsed as strings and are supposed to be one of supported GMPEs. See list at :class:`GMPELogicTree`. sourceModel Branches contain references to different PSHA source models. Values are treated as file names, relatively to base path. maxMagGRRelative Different values to add to Gutenberg-Richter ("GR") maximum magnitude. Value should be interpretable as float. bGRRelative Values to add to GR "b" value. Parsed as float. maxMagGRAbsolute Values to replace GR maximum magnitude. Values expected to be lists of floats separated by space, one float for each GR MFD in a target source in order of appearance. abGRAbsolute Values to replace "a" and "b" values of GR MFD. Lists of pairs of floats, one pair for one GR MFD in a target source. incrementalMFDAbsolute Replaces an evenly discretized MFD with the values provided simpleFaultDipRelative Increases or decreases the angle of fault dip from that given in the original source model simpleFaultDipAbsolute Replaces the fault dip in the specified source(s) simpleFaultGeometryAbsolute Replaces the simple fault geometry (trace, upper seismogenic depth lower seismogenic depth and dip) of a given source with the values provided complexFaultGeometryAbsolute Replaces the complex fault geometry edges of a given source with the values provided characteristicFaultGeometryAbsolute Replaces the complex fault geometry surface of a given source with the values provided :param filters: Dictionary, a set of filters to specify which sources should the uncertainty be applied to. Represented as branchset element's attributes in xml: applyToSources The uncertainty should be applied only to specific sources. This filter is required for absolute uncertainties (also only one source can be used for those). Value should be the list of source ids. Can be used only in source model logic tree. applyToSourceType Can be used in the source model logic tree definition. Allows to specify to which source type (area, point, simple fault, complex fault) the uncertainty applies to. applyToTectonicRegionType Can be used in both the source model and GMPE logic trees. Allows to specify to which tectonic region type (Active Shallow Crust, Stable Shallow Crust, etc.) the uncertainty applies to. This filter is required for all branchsets in GMPE logic tree. """ def __init__(self, uncertainty_type, filters): self.branches = [] self.uncertainty_type = uncertainty_type self.filters = filters
[docs] def enumerate_paths(self): """ Generate all possible paths starting from this branch set. :returns: Generator of two-item tuples. Each tuple contains weight of the path (calculated as a product of the weights of all path's branches) and list of path's :class:`Branch` objects. Total sum of all paths' weights is 1.0 """ for path in self._enumerate_paths([]): flat_path = [] weight = 1.0 while path: path, branch = path weight *= branch.weight flat_path.append(branch) yield weight, flat_path[::-1]
def _enumerate_paths(self, prefix_path): """ Recursive (private) part of :func:`enumerate_paths`. Returns generator of recursive lists of two items, where second item is the branch object and first one is itself list of two items. """ for branch in self.branches: path = [prefix_path, branch] if branch.child_branchset is not None: for subpath in branch.child_branchset._enumerate_paths(path): yield subpath else: yield path
[docs] def get_branch_by_id(self, branch_id): """ Return :class:`Branch` object belonging to this branch set with id equal to ``branch_id``. """ for branch in self.branches: if branch.branch_id == branch_id: return branch raise AssertionError("couldn't find branch '%s'" % branch_id)
[docs] def filter_source(self, source): # pylint: disable=R0911,R0912 """ Apply filters to ``source`` and return ``True`` if uncertainty should be applied to it. """ for key, value in self.filters.items(): if key == 'applyToTectonicRegionType': if value != source.tectonic_region_type: return False elif key == 'applyToSourceType': if value == 'area': if not isinstance(source, ohs.AreaSource): return False elif value == 'point': # area source extends point source if (not isinstance(source, ohs.PointSource) or isinstance(source, ohs.AreaSource)): return False elif value == 'simpleFault': if not isinstance(source, ohs.SimpleFaultSource): return False elif value == 'complexFault': if not isinstance(source, ohs.ComplexFaultSource): return False elif value == 'characteristicFault': if not isinstance(source, ohs.CharacteristicFaultSource): return False else: raise AssertionError("unknown source type '%s'" % value) elif key == 'applyToSources': if source and source.source_id not in value: return False else: raise AssertionError("unknown filter '%s'" % key) # All filters pass, return True. return True
[docs] def apply_uncertainty(self, value, source): """ Apply this branchset's uncertainty with value ``value`` to source ``source``, if it passes :meth:`filters <filter_source>`. This method is not called for uncertainties of types "gmpeModel" and "sourceModel". :param value: The actual uncertainty value of :meth:`sampled <sample>` branch. Type depends on uncertainty type. :param source: The opensha source data object. :return: 0 if the source was not changed, 1 otherwise """ if not self.filter_source(source): # source didn't pass the filter return 0 if self.uncertainty_type in MFD_UNCERTAINTY_TYPES: self._apply_uncertainty_to_mfd(source.mfd, value) elif self.uncertainty_type in GEOMETRY_UNCERTAINTY_TYPES: self._apply_uncertainty_to_geometry(source, value) else: raise AssertionError("unknown uncertainty type '%s'" % self.uncertainty_type) return 1
def _apply_uncertainty_to_geometry(self, source, value): """ Modify ``source`` geometry with the uncertainty value ``value`` """ if self.uncertainty_type == 'simpleFaultDipRelative': source.modify('adjust_dip', dict(increment=value)) elif self.uncertainty_type == 'simpleFaultDipAbsolute': source.modify('set_dip', dict(dip=value)) elif self.uncertainty_type == 'simpleFaultGeometryAbsolute': trace, usd, lsd, dip, spacing = value source.modify( 'set_geometry', dict(fault_trace=trace, upper_seismogenic_depth=usd, lower_seismogenic_depth=lsd, dip=dip, spacing=spacing)) elif self.uncertainty_type == 'complexFaultGeometryAbsolute': edges, spacing = value source.modify('set_geometry', dict(edges=edges, spacing=spacing)) elif self.uncertainty_type == 'characteristicFaultGeometryAbsolute': source.modify('set_geometry', dict(surface=value)) def _apply_uncertainty_to_mfd(self, mfd, value): """ Modify ``mfd`` object with uncertainty value ``value``. """ if self.uncertainty_type == 'abGRAbsolute': a, b = value mfd.modify('set_ab', dict(a_val=a, b_val=b)) elif self.uncertainty_type == 'bGRRelative': mfd.modify('increment_b', dict(value=value)) elif self.uncertainty_type == 'maxMagGRRelative': mfd.modify('increment_max_mag', dict(value=value)) elif self.uncertainty_type == 'maxMagGRAbsolute': mfd.modify('set_max_mag', dict(value=value)) elif self.uncertainty_type == 'incrementalMFDAbsolute': min_mag, bin_width, occur_rates = value mfd.modify('set_mfd', dict(min_mag=min_mag, bin_width=bin_width, occurrence_rates=occur_rates)) def __repr__(self): return repr(self.branches)
def _bsnodes(fname, branchinglevel): if branchinglevel.tag.endswith('logicTreeBranchingLevel'): if len(branchinglevel) > 1: raise InvalidLogicTree( '%s: Branching level %s has multiple branchsets' % (fname, branchinglevel['branchingLevelID'])) return branchinglevel.nodes elif branchinglevel.tag.endswith('logicTreeBranchSet'): return [branchinglevel] else: raise ValueError('Expected BranchingLevel/BranchSet, got %s' % branchinglevel)
[docs]class FakeSmlt(object): """ A replacement for the SourceModelLogicTree class, to be used when there is a trivial source model logic tree. In practice, when `source_model_logic_tree_file` is missing but there is a `source_model_file` in the job.ini file. """ def __init__(self, filename, seed=0, num_samples=0): self.filename = filename self.basepath = os.path.dirname(filename) self.seed = seed self.num_samples = num_samples self.on_each_source = False self.num_paths = 1 with open(self.filename, encoding='utf-8') as f: xml = self.tectonic_region_type = set(TRT_REGEX.findall(xml))
[docs] def gen_source_models(self, gsim_lt): """ Yield the underlying LtSourceModel, multiple times if there is sampling """ num_gsim_paths = 1 if self.num_samples else gsim_lt.get_num_paths() for i, rlz in enumerate(self): yield LtSourceModel( rlz.value, rlz.weight, ('b1',), [], num_gsim_paths, i, 1)
[docs] def apply_uncertainties(self, branch_ids, sourcegroup): """ :returns: the sourcegroup unchanged """ return sourcegroup
def __iter__(self): name = os.path.basename(self.filename) smlt_path = ('b1',) if self.num_samples: # many realizations of equal weight weight = 1. / self.num_samples for i in range(self.num_samples): yield Realization(name, weight, smlt_path, None, smlt_path) else: # there is a single realization yield Realization(name, 1.0, smlt_path, 0, smlt_path)
Info = collections.namedtuple('Info', 'smpaths, applytosources')
[docs]def collect_info(smlt): """ Given a path to a source model logic tree, collect all of the path names to the source models it contains and build 1. a dictionary source model branch ID -> paths 2. a dictionary source model branch ID -> source IDs in applyToSources :param smlt: source model logic tree file :returns: an Info namedtupled containing the two dictionaries """ n = try: blevels = n.logicTree except Exception: raise InvalidFile('%s is not a valid source_model_logic_tree_file' % smlt) paths = collections.defaultdict(set) # branchID -> paths applytosources = collections.defaultdict(list) # branchID -> source IDs for blevel in blevels: for bset in _bsnodes(smlt, blevel): if 'applyToSources' in bset.attrib: applytosources[bset.get('applyToBranches')].extend( bset['applyToSources'].split()) for br in bset: with node.context(smlt, br): fnames = unique(br.uncertaintyModel.text.split()) paths[br['branchID']].update(get_paths(smlt, fnames)) return Info({k: sorted(v) for k, v in paths.items()}, applytosources)
[docs]def get_paths(smlt, fnames): base_path = os.path.dirname(smlt) paths = [] for fname in fnames: if os.path.isabs(fname): raise InvalidFile('%s: %s must be a relative path' % (smlt, fname)) fname = os.path.abspath(os.path.join(base_path, fname)) if os.path.exists(fname): # consider only real paths paths.append(fname) return paths
[docs]def read_source_groups(fname): """ :param fname: a path to a source model XML file :return: a list of SourceGroup objects containing source nodes """ smodel = src_groups = [] if smodel[0].tag.endswith('sourceGroup'): # NRML 0.5 format for sg_node in smodel: sg = SourceGroup(sg_node['tectonicRegion']) sg.sources = sg_node.nodes src_groups.append(sg) else: # NRML 0.4 format: smodel is a list of source nodes src_groups.extend(SourceGroup.collect(smodel)) return src_groups
[docs]class SourceModelLogicTree(object): """ Source model logic tree parser. :param filename: Full pathname of logic tree file :param validate: Boolean indicating whether or not the tree should be validated while parsed. This should be set to ``True`` on initial load of the logic tree (before importing it to the database) and to ``False`` on workers side (when loaded from the database). :raises LogicTreeError: If logic tree file has a logic error, which can not be prevented by xml schema rules (like referencing sources with missing id). """ _xmlschema = None FILTERS = ('applyToTectonicRegionType', 'applyToSources', 'applyToSourceType') def __init__(self, filename, validate=True, seed=0, num_samples=0): self.filename = filename self.basepath = os.path.dirname(filename) self.seed = seed self.num_samples = num_samples self.branches = {} # branch_id -> branch self.bsetdict = {} self.open_ends = set() self.tectonic_region_types = set() self.source_types = set() self.root_branchset = None root = try: tree = root.logicTree except AttributeError: raise LogicTreeError( root, self.filename, "missing logicTree node") self.parse_tree(tree, validate) @property def on_each_source(self): """ True if there is an applyToSources for each source. """ return ( and == self.source_ids)
[docs] def parse_tree(self, tree_node, validate): """ Parse the whole tree and point ``root_branchset`` attribute to the tree's root. """ = collect_info(self.filename) self.source_ids = collections.defaultdict(list) t0 = time.time() for depth, branchinglevel_node in enumerate(tree_node.nodes): self.parse_branchinglevel(branchinglevel_node, depth, validate) dt = time.time() - t0 if validate: bname = os.path.basename(self.filename)'Validated %s in %.2f seconds', bname, dt)
[docs] def parse_branchinglevel(self, branchinglevel_node, depth, validate): """ Parse one branching level. :param branchinglevel_node: ``etree.Element`` object with tag "logicTreeBranchingLevel". :param depth: The sequential number of this branching level, based on 0. :param validate: Whether or not the branching level, its branchsets and their branches should be validated. Enumerates children branchsets and call :meth:`parse_branchset`, :meth:`validate_branchset`, :meth:`parse_branches` and finally :meth:`apply_branchset` for each. Keeps track of "open ends" -- the set of branches that don't have any child branchset on this step of execution. After processing of every branching level only those branches that are listed in it can have child branchsets (if there is one on the next level). """ new_open_ends = set() for number, branchset_node in enumerate( _bsnodes(self.filename, branchinglevel_node)): attrs = branchset_node.attrib.copy() self.bsetdict[attrs.pop('branchSetID')] = attrs branchset = self.parse_branchset(branchset_node, depth, number, validate) self.parse_branches(branchset_node, branchset, validate) if self.root_branchset is None: # not set yet self.num_paths = 1 self.root_branchset = branchset else: self.apply_branchset(branchset_node, branchset) for branch in branchset.branches: new_open_ends.add(branch) self.num_paths *= len(branchset.branches) if number > 0: raise InvalidLogicTree('there is a branching level with multiple' ' branchsets in %s' % self.filename) self.open_ends.clear() self.open_ends.update(new_open_ends)
[docs] def parse_branchset(self, branchset_node, depth, number, validate): """ Create :class:`BranchSet` object using data in ``branchset_node``. :param branchset_node: ``etree.Element`` object with tag "logicTreeBranchSet". :param depth: The sequential number of branchset's branching level, based on 0. :param number: Index number of this branchset inside branching level, based on 0. :param validate: Whether or not filters defined in branchset and the branchset itself should be validated. :returns: An instance of :class:`BranchSet` with filters applied but with no branches (they're attached in :meth:`parse_branches`). """ uncertainty_type = branchset_node.attrib.get('uncertaintyType') filters = dict((filtername, branchset_node.attrib.get(filtername)) for filtername in self.FILTERS if filtername in branchset_node.attrib) if validate: self.validate_filters(branchset_node, uncertainty_type, filters) filters = self.parse_filters(branchset_node, uncertainty_type, filters) branchset = BranchSet(uncertainty_type, filters) if validate: self.validate_branchset(branchset_node, depth, number, branchset) return branchset
[docs] def parse_branches(self, branchset_node, branchset, validate): """ Create and attach branches at ``branchset_node`` to ``branchset``. :param branchset_node: Same as for :meth:`parse_branchset`. :param branchset: An instance of :class:`BranchSet`. :param validate: Whether or not branches' uncertainty values should be validated. Checks that each branch has :meth:`valid <validate_uncertainty_value>` value, unique id and that all branches have total weight of 1.0. :return: ``None``, all branches are attached to provided branchset. """ bs_id = branchset_node['branchSetID'] weight_sum = 0 branches = branchset_node.nodes values = [] for branchnode in branches: weight = ~branchnode.uncertaintyWeight weight_sum += weight value_node = node_from_elem(branchnode.uncertaintyModel) if value_node.text is not None: values.append(value_node.text.strip()) if validate: self.validate_uncertainty_value( value_node, branchnode, branchset) value = self.parse_uncertainty_value(value_node, branchset) branch_id = branchnode.attrib.get('branchID') branch = Branch(bs_id, branch_id, weight, value) if branch_id in self.branches: raise LogicTreeError( branchnode, self.filename, "branchID '%s' is not unique" % branch_id) self.branches[branch_id] = branch branchset.branches.append(branch) if abs(weight_sum - 1.0) > pmf.PRECISION: raise LogicTreeError( branchset_node, self.filename, "branchset weights don't sum up to 1.0") if len(set(values)) < len(values): raise LogicTreeError( branchset_node, self.filename, "there are duplicate values in uncertaintyModel: " + ' '.join(values))
[docs] def gen_source_models(self, gsim_lt): """ Yield empty LtSourceModel instances (one per effective realization) """ samples_by_lt_path = self.samples_by_lt_path() for i, rlz in enumerate(get_effective_rlzs(self)): smpath = rlz.lt_path num_samples = samples_by_lt_path[smpath] num_gsim_paths = (num_samples if self.num_samples else gsim_lt.get_num_paths()) yield LtSourceModel( rlz.value, rlz.weight / num_samples, smpath, [], num_gsim_paths, i, num_samples)
[docs] def sample_path(self, seed): """ Return the model name and a list of branch ids. :param seed: the seed used for the sampling """ branchset = self.root_branchset branch_ids = [] while branchset is not None: [branch] = sample(branchset.branches, 1, seed) branch_ids.append(branch.branch_id) branchset = branch.child_branchset modelname = self.root_branchset.get_branch_by_id(branch_ids[0]).value return modelname, branch_ids
def __iter__(self): """ Yield Realization tuples. Notice that the weight is homogeneous when sampling is enabled, since it is accounted for in the sampling procedure. """ if self.num_samples: # random sampling of the logic tree weight = 1. / self.num_samples for i in range(self.num_samples): name, sm_lt_path = self.sample_path(self.seed + i) yield Realization(name, weight, tuple(sm_lt_path), None, tuple(sm_lt_path)) else: # full enumeration ordinal = 0 for weight, smlt_path in self.root_branchset.enumerate_paths(): name = smlt_path[0].value smlt_branch_ids = [branch.branch_id for branch in smlt_path] yield Realization(name, weight, tuple(smlt_branch_ids), ordinal, tuple(smlt_branch_ids)) ordinal += 1
[docs] def parse_uncertainty_value(self, node, branchset): """ See superclass' method for description and signature specification. Doesn't change source model file name, converts other values to either pair of floats or a single float depending on uncertainty type. """ if branchset.uncertainty_type == 'sourceModel': return node.text.strip() elif branchset.uncertainty_type == 'abGRAbsolute': [a, b] = node.text.strip().split() return float(a), float(b) elif branchset.uncertainty_type == 'incrementalMFDAbsolute': min_mag, bin_width = (node.incrementalMFD["minMag"], node.incrementalMFD["binWidth"]) return min_mag, bin_width, ~node.incrementalMFD.occurRates elif branchset.uncertainty_type == 'simpleFaultGeometryAbsolute': return self._parse_simple_fault_geometry_surface( node.simpleFaultGeometry) elif branchset.uncertainty_type == 'complexFaultGeometryAbsolute': return self._parse_complex_fault_geometry_surface( node.complexFaultGeometry) elif branchset.uncertainty_type ==\ 'characteristicFaultGeometryAbsolute': surfaces = [] for geom_node in node.surface: if "simpleFaultGeometry" in geom_node.tag: trace, usd, lsd, dip, spacing =\ self._parse_simple_fault_geometry_surface(geom_node) surfaces.append(geo.SimpleFaultSurface.from_fault_data( trace, usd, lsd, dip, spacing)) elif "complexFaultGeometry" in geom_node.tag: edges, spacing =\ self._parse_complex_fault_geometry_surface(geom_node) surfaces.append(geo.ComplexFaultSurface.from_fault_data( edges, spacing)) elif "planarSurface" in geom_node.tag: surfaces.append( self._parse_planar_geometry_surface(geom_node)) else: pass if len(surfaces) > 1: return geo.MultiSurface(surfaces) else: return surfaces[0] else: return float(node.text.strip())
def _parse_simple_fault_geometry_surface(self, node): """ Parses a simple fault geometry surface """ spacing = node["spacing"] usd, lsd, dip = (~node.upperSeismoDepth, ~node.lowerSeismoDepth, ~node.dip) # Parse the geometry coords = split_coords_2d(~node.LineString.posList) trace = geo.Line([geo.Point(*p) for p in coords]) return trace, usd, lsd, dip, spacing def _parse_complex_fault_geometry_surface(self, node): """ Parses a complex fault geometry surface """ spacing = node["spacing"] edges = [] for edge_node in node.nodes: coords = split_coords_3d(~edge_node.LineString.posList) edges.append(geo.Line([geo.Point(*p) for p in coords])) return edges, spacing def _parse_planar_geometry_surface(self, node): """ Parses a planar geometry surface """ nodes = [] for key in ["topLeft", "topRight", "bottomRight", "bottomLeft"]: nodes.append(geo.Point(getattr(node, key)["lon"], getattr(node, key)["lat"], getattr(node, key)["depth"])) top_left, top_right, bottom_right, bottom_left = tuple(nodes) return geo.PlanarSurface.from_corner_points( top_left, top_right, bottom_right, bottom_left)
[docs] def validate_uncertainty_value(self, node, branchnode, branchset): """ See superclass' method for description and signature specification. Checks that the following conditions are met: * For uncertainty of type "sourceModel": referenced file must exist and be readable. This is checked in :meth:`collect_source_model_data` along with saving the source model information. * For uncertainty of type "abGRAbsolute": value should be two float values. * For both absolute uncertainties: the source (only one) must be referenced in branchset's filter "applyToSources". * For all other cases: value should be a single float value. """ _float_re = re.compile(r'^(\+|\-)?(\d+|\d*\.\d+)$') if branchset.uncertainty_type == 'sourceModel': try: for fname in node.text.strip().split(): self.collect_source_model_data( branchnode['branchID'], fname) except Exception as exc: raise LogicTreeError(node, self.filename, str(exc)) from exc elif branchset.uncertainty_type == 'abGRAbsolute': ab = (node.text.strip()).split() if len(ab) == 2: a, b = ab if _float_re.match(a) and _float_re.match(b): return raise LogicTreeError( node, self.filename, 'expected a pair of floats separated by space') elif branchset.uncertainty_type == 'incrementalMFDAbsolute': pass elif branchset.uncertainty_type == 'simpleFaultGeometryAbsolute': self._validate_simple_fault_geometry(node.simpleFaultGeometry, _float_re) elif branchset.uncertainty_type == 'complexFaultGeometryAbsolute': self._validate_complex_fault_geometry(node.complexFaultGeometry, _float_re) elif branchset.uncertainty_type ==\ 'characteristicFaultGeometryAbsolute': for geom_node in node.surface: if "simpleFaultGeometry" in geom_node.tag: self._validate_simple_fault_geometry(geom_node, _float_re) elif "complexFaultGeometry" in geom_node.tag: self._validate_complex_fault_geometry(geom_node, _float_re) elif "planarSurface" in geom_node.tag: self._validate_planar_fault_geometry(geom_node, _float_re) else: raise LogicTreeError( geom_node, self.filename, "Surface geometry type not recognised") else: try: float(node.text) except (TypeError, ValueError): raise LogicTreeError( node, self.filename, 'expected single float value')
def _validate_simple_fault_geometry(self, node, _float_re): """ Validates a node representation of a simple fault geometry """ try: # Parse the geometry coords = split_coords_2d(~node.LineString.posList) trace = geo.Line([geo.Point(*p) for p in coords]) except ValueError: # If the geometry cannot be created then use the LogicTreeError # to point the user to the incorrect node. Hence, if trace is # compiled successfully then len(trace) is True, otherwise it is # False trace = [] if len(trace): return raise LogicTreeError( node, self.filename, "'simpleFaultGeometry' node is not valid") def _validate_complex_fault_geometry(self, node, _float_re): """ Validates a node representation of a complex fault geometry - this check merely verifies that the format is correct. If the geometry does not conform to the Aki & Richards convention this will not be verified here, but will raise an error when the surface is created. """ valid_edges = [] for edge_node in node.nodes: try: coords = split_coords_3d(edge_node.LineString.posList.text) edge = geo.Line([geo.Point(*p) for p in coords]) except ValueError: # See use of validation error in simple geometry case # The node is valid if all of the edges compile correctly edge = [] if len(edge): valid_edges.append(True) else: valid_edges.append(False) if node["spacing"] and all(valid_edges): return raise LogicTreeError( node, self.filename, "'complexFaultGeometry' node is not valid") def _validate_planar_fault_geometry(self, node, _float_re): """ Validares a node representation of a planar fault geometry """ valid_spacing = node["spacing"] for key in ["topLeft", "topRight", "bottomLeft", "bottomRight"]: lon = getattr(node, key)["lon"] lat = getattr(node, key)["lat"] depth = getattr(node, key)["depth"] valid_lon = (lon >= -180.0) and (lon <= 180.0) valid_lat = (lat >= -90.0) and (lat <= 90.0) valid_depth = (depth >= 0.0) is_valid = valid_lon and valid_lat and valid_depth if not is_valid or not valid_spacing: raise LogicTreeError( node, self.filename, "'planarFaultGeometry' node is not valid")
[docs] def parse_filters(self, branchset_node, uncertainty_type, filters): """ See superclass' method for description and signature specification. Converts "applyToSources" filter value by just splitting it to a list. """ if 'applyToSources' in filters: filters['applyToSources'] = filters['applyToSources'].split() return filters
[docs] def validate_filters(self, branchset_node, uncertainty_type, filters): """ See superclass' method for description and signature specification. Checks that the following conditions are met: * "sourceModel" uncertainties can not have filters. * Absolute uncertainties must have only one filter -- "applyToSources", with only one source id. * All other uncertainty types can have either no or one filter. * Filter "applyToSources" must mention only source ids that exist in source models. * Filter "applyToTectonicRegionType" must mention only tectonic region types that exist in source models. * Filter "applyToSourceType" must mention only source types that exist in source models. """ if uncertainty_type == 'sourceModel' and filters: raise LogicTreeError( branchset_node, self.filename, 'filters are not allowed on source model uncertainty') if len(filters) > 1: raise LogicTreeError( branchset_node, self.filename, "only one filter is allowed per branchset") if 'applyToTectonicRegionType' in filters: if not filters['applyToTectonicRegionType'] \ in self.tectonic_region_types: raise LogicTreeError( branchset_node, self.filename, "source models don't define sources of tectonic region " "type '%s'" % filters['applyToTectonicRegionType']) if uncertainty_type in ('abGRAbsolute', 'maxMagGRAbsolute', 'simpleFaultGeometryAbsolute', 'complexFaultGeometryAbsolute'): if not filters or not list(filters) == ['applyToSources'] \ or not len(filters['applyToSources'].split()) == 1: raise LogicTreeError( branchset_node, self.filename, "uncertainty of type '%s' must define 'applyToSources' " "with only one source id" % uncertainty_type) if uncertainty_type in ('simpleFaultDipRelative', 'simpleFaultDipAbsolute'): if not filters or (not ('applyToSources' in filters.keys()) and not ('applyToSourceType' in filters.keys())): raise LogicTreeError( branchset_node, self.filename, "uncertainty of type '%s' must define either" "'applyToSources' or 'applyToSourceType'" % uncertainty_type) if 'applyToSourceType' in filters: if not filters['applyToSourceType'] in self.source_types: raise LogicTreeError( branchset_node, self.filename, "source models don't define sources of type '%s'" % filters['applyToSourceType']) if 'applyToSources' in filters: if (len(self.source_ids) > 1 and 'applyToBranches' not in branchset_node.attrib): raise LogicTreeError( branchset_node, self.filename, "applyToBranch must be " "specified together with applyToSources") cnt = 0 for source_id in filters['applyToSources'].split(): for source_ids in self.source_ids.values(): if source_id in source_ids: cnt += 1 if cnt == 0: raise LogicTreeError( branchset_node, self.filename, "source with id '%s' is not defined in source " "models" % source_id)
[docs] def validate_branchset(self, branchset_node, depth, number, branchset): """ See superclass' method for description and signature specification. Checks that the following conditions are met: * First branching level must contain exactly one branchset, which must be of type "sourceModel". * All other branchsets must not be of type "sourceModel" or "gmpeModel". """ if depth == 0: if branchset.uncertainty_type != 'sourceModel': raise LogicTreeError( branchset_node, self.filename, 'first branchset must define an uncertainty ' 'of type "sourceModel"') else: if branchset.uncertainty_type == 'sourceModel': raise LogicTreeError( branchset_node, self.filename, 'uncertainty of type "sourceModel" can be defined ' 'on first branchset only') elif branchset.uncertainty_type == 'gmpeModel': raise LogicTreeError( branchset_node, self.filename, 'uncertainty of type "gmpeModel" is not allowed ' 'in source model logic tree')
[docs] def apply_branchset(self, branchset_node, branchset): """ See superclass' method for description and signature specification. Parses branchset node's attribute ``@applyToBranches`` to apply following branchests to preceding branches selectively. Branching level can have more than one branchset exactly for this: different branchsets can apply to different open ends. Checks that branchset tries to be applied only to branches on previous branching level which do not have a child branchset yet. """ apply_to_branches = branchset_node.attrib.get('applyToBranches') if apply_to_branches: apply_to_branches = apply_to_branches.split() for branch_id in apply_to_branches: if branch_id not in self.branches: raise LogicTreeError( branchset_node, self.filename, "branch '%s' is not yet defined" % branch_id) branch = self.branches[branch_id] if branch.child_branchset is not None: raise LogicTreeError( branchset_node, self.filename, "branch '%s' already has child branchset" % branch_id) branch.child_branchset = branchset else: for branch in self.open_ends: branch.child_branchset = branchset
def _get_source_model(self, source_model_file): return open(os.path.join(self.basepath, source_model_file), encoding='utf-8')
[docs] def collect_source_model_data(self, branch_id, source_model): """ Parse source model file and collect information about source ids, source types and tectonic region types available in it. That information is used then for :meth:`validate_filters` and :meth:`validate_uncertainty_value`. """ # using regular expressions is a lot faster than using the with self._get_source_model(source_model) as sm: xml = self.tectonic_region_types.update(TRT_REGEX.findall(xml)) self.source_ids[branch_id].extend(ID_REGEX.findall(xml)) self.source_types.update(SOURCE_TYPE_REGEX.findall(xml))
[docs] def apply_uncertainties(self, branch_ids, source_group): """ Parse the path through the source model logic tree and return "apply uncertainties" function. :param branch_ids: List of string identifiers of branches, representing the path through source model logic tree. :param source_group: A group of sources :return: A copy of the original group with modified sources """ branchset = self.root_branchset branchsets_and_uncertainties = [] branch_ids = list(branch_ids[::-1]) while branchset is not None: branch = branchset.get_branch_by_id(branch_ids.pop(-1)) if not branchset.uncertainty_type == 'sourceModel': branchsets_and_uncertainties.append((branchset, branch.value)) branchset = branch.child_branchset if not branchsets_and_uncertainties: return source_group # nothing changed sg = copy.deepcopy(source_group) sg.applied_uncertainties = [] sg.changed = numpy.zeros(len(sg.sources), int) for branchset, value in branchsets_and_uncertainties: for s, source in enumerate(sg.sources): changed = branchset.apply_uncertainty(value, source) if changed: sg.changed[s] += changed sg.applied_uncertainties.append( (branchset.uncertainty_type, value)) return sg # something changed
[docs] def samples_by_lt_path(self): """ Returns a dictionary lt_path -> how many times that path was sampled """ return collections.Counter(rlz.lt_path for rlz in self)
def __toh5__(self): tbl = [] for brid, br in self.branches.items(): tbl.append((br.bs_id, brid, br.value, br.weight)) dt = [('branchset', hdf5.vstr), ('branch', hdf5.vstr), ('uncertainty', hdf5.vstr), ('weight', float)] dic = dict(branches=numpy.array(tbl, dt), branchsets=toml.dumps(self.bsetdict)) return dic, {} def __fromh5__(self, dic, attrs): # TODO: this is not complete the child_branchset must be built too self.bsetdict = toml.loads(dic['branchsets'][()]) self.branches = {} self.root_branchset = BranchSet('sourceModel', {}) for rec in dic['branches']: bs = rec['branchset'] dic = self.bsetdict[bs] br = Branch(bs, rec['branch'], rec['weight'], rec['uncertainty']) self.branches[br.branch_id] = br apply_to_branches = dic.get('applyToBranches') if apply_to_branches: for br_id in apply_to_branches.split(): br.child_branchset = self.branches[br_id] def __str__(self): return '<%s%s>' % (self.__class__.__name__, repr(self.root_branchset))
# used in GsimLogicTree BranchTuple = namedtuple('BranchTuple', 'trt id gsim weight effective')
[docs]class InvalidLogicTree(Exception): pass
[docs]class ImtWeight(object): """ A composite weight by IMTs extracted from the gsim_logic_tree_file """ def __init__(self, branch, fname): with context(fname, branch.uncertaintyWeight): nodes = list(branch.getnodes('uncertaintyWeight')) if 'imt' in nodes[0].attrib: raise InvalidLogicTree('The first uncertaintyWeight has an imt' ' attribute') self.dic = {'weight': float(nodes[0].text)} imts = [] for n in nodes[1:]: self.dic[n['imt']] = float(n.text) imts.append(n['imt']) if len(set(imts)) < len(imts): raise InvalidLogicTree( 'There are duplicated IMTs in the weights') def __mul__(self, other): new = object.__new__(self.__class__) if isinstance(other, self.__class__): keys = set(self.dic) | set(other.dic) new.dic = {k: self[k] * other[k] for k in keys} else: # assume a float new.dic = {k: self.dic[k] * other for k in self.dic} return new __rmul__ = __mul__ def __add__(self, other): new = object.__new__(self.__class__) if isinstance(other, self.__class__): new.dic = {k: self.dic[k] + other[k] for k in self.dic} else: # assume a float new.dic = {k: self.dic[k] + other for k in self.dic} return new __radd__ = __add__ def __truediv__(self, other): new = object.__new__(self.__class__) if isinstance(other, self.__class__): new.dic = {k: self.dic[k] / other[k] for k in self.dic} else: # assume a float new.dic = {k: self.dic[k] / other for k in self.dic} return new
[docs] def is_one(self): """ Check that all the inner weights are 1 up to the precision """ return all(abs(v - 1.) < pmf.PRECISION for v in self.dic.values() if v)
def __getitem__(self, imt): try: return self.dic[imt] except KeyError: return self.dic['weight'] def __repr__(self): return '<%s %s>' % (self.__class__.__name__, self.dic)
[docs]def to_toml(uncertainty): """ Converts an uncertainty node into a TOML string """ text = uncertainty.text.strip() if not text.startswith('['): # a bare GSIM name was passed text = '[%s]' % text for k, v in uncertainty.attrib.items(): try: v = ast.literal_eval(v) except (SyntaxError, ValueError): v = repr(v) text += '\n%s = %s' % (k, v) return text
[docs]class GsimLogicTree(object): """ A GsimLogicTree instance is an iterable yielding `Realization` tuples with attributes `value`, `weight` and `lt_path`, where `value` is a dictionary {trt: gsim}, `weight` is a number in the interval 0..1 and `lt_path` is a tuple with the branch ids of the given realization. :param str fname: full path of the gsim_logic_tree file :param tectonic_region_types: a sequence of distinct tectonic region types :param ltnode: usually None, but it can also be a :class:`openquake.hazardlib.nrml.Node` object describing the GSIM logic tree XML file, to avoid reparsing it """
[docs] @classmethod def from_(cls, gsim): """ Generate a trivial GsimLogicTree from a single GSIM instance. """ ltbranch = N('logicTreeBranch', {'branchID': 'b1'}, nodes=[N('uncertaintyModel', text=str(gsim)), N('uncertaintyWeight', text='1.0')]) lt = N('logicTree', {'logicTreeID': 'lt1'}, nodes=[N('logicTreeBranchingLevel', {'branchingLevelID': 'bl1'}, nodes=[N('logicTreeBranchSet', {'applyToTectonicRegionType': '*', 'branchSetID': 'bs1', 'uncertaintyType': 'gmpeModel'}, nodes=[ltbranch])])]) return cls(repr(gsim), ['*'], ltnode=lt)
def __init__(self, fname, tectonic_region_types=['*'], ltnode=None): self.filename = fname trts = sorted(tectonic_region_types) if len(trts) > len(set(trts)): raise ValueError( 'The given tectonic region types are not distinct: %s' % ','.join(trts)) self.values = collections.defaultdict(list) # {trt: gsims} self._ltnode = ltnode or self.gmpe_tables = set() # populated right below self.branches = self._build_trts_branches(trts) if tectonic_region_types and not self.branches: raise InvalidLogicTree( 'Could not find branches with attribute ' "'applyToTectonicRegionType' in %s" % set(tectonic_region_types)) @property def req_site_params(self): site_params = set() for trt in self.values: for gsim in self.values[trt]: site_params.update(gsim.REQUIRES_SITES_PARAMETERS) return site_params
[docs] def check_imts(self, imts): """ Make sure the IMTs are recognized by all GSIMs in the logic tree """ for trt in self.values: for gsim in self.values[trt]: for attr in dir(gsim): coeffs = getattr(gsim, attr) if not isinstance(coeffs, CoeffsTable): continue for imt in imts: if imt.startswith('SA'): try: coeffs[from_string(imt)] except KeyError: raise ValueError( '%s is out of the period range defined ' 'for %s' % (imt, gsim))
def __toh5__(self): weights = set() for branch in self.branches: weights.update(branch.weight.dic) dt = [('trt', hdf5.vstr), ('branch', hdf5.vstr), ('uncertainty', hdf5.vstr)] + [ (weight, float) for weight in sorted(weights)] branches = [(b.trt,, b.gsim) + tuple(b.weight[weight] for weight in sorted(weights)) for b in self.branches] dic = {'branches': numpy.array(branches, dt)} # manage gmpe_tables, if any if hasattr(self, 'filename'): # missing for fake logic trees dirname = os.path.dirname(self.filename) for gmpe_table in sorted(self.gmpe_tables): dic[os.path.basename(gmpe_table)] = d = {} with hdf5.File(os.path.join(dirname, gmpe_table), 'r') as f: for group, dset in f.items(): if hasattr(dset, 'shape'): # dataset, not group d[group] = dset[()] if group == 'Distances': d['distance_type'] = ( python3compat.decode(dset.attrs['metric'])) else: d[group] = {k: ds[()] for k, ds in dset.items()} return dic, {} def __fromh5__(self, dic, attrs): self.branches = [] self.values = collections.defaultdict(list) for branch in dic.pop('branches'): if 'id' in branch.dtype.names: # engine < 3.6 br_id = branch['id'] gsim_ = branch['gsim'] else: br_id = branch['branch'] gsim_ = branch['uncertainty'] gsim = valid.gsim(gsim_) self.values[branch['trt']].append(gsim) weight = object.__new__(ImtWeight) # branch has dtype ('trt', 'branch', 'uncertainty', 'weight', ...) weight.dic = {w: branch[w] for w in branch.dtype.names[3:]} if len(weight.dic) > 1: gsim.weight = weight bt = BranchTuple(branch['trt'], br_id, gsim, weight, True) self.branches.append(bt)
[docs] def reduce(self, trts): """ Reduce the GsimLogicTree. :param trts: a subset of tectonic region types :returns: a reduced GsimLogicTree instance """ new = object.__new__(self.__class__) vars(new).update(vars(self)) if trts != {'*'}: new.branches = [] for br in self.branches: branch = BranchTuple(br.trt,, br.gsim, br.weight, br.trt in trts) new.branches.append(branch) return new
[docs] def get_num_branches(self): """ Return the number of effective branches for tectonic region type, as a dictionary. """ num = {} for trt, branches in itertools.groupby( self.branches, operator.attrgetter('trt')): num[trt] = sum(1 for br in branches if br.effective) return num
[docs] def get_num_paths(self): """ Return the effective number of paths in the tree. """ # NB: the algorithm assume a symmetric logic tree for the GSIMs; # in the future we may relax such assumption num_branches = self.get_num_branches() if not sum(num_branches.values()): return 0 num = 1 for val in num_branches.values(): if val: # the branch is effective num *= val return num
def _build_trts_branches(self, tectonic_region_types): # do the parsing, called at instantiation time to populate .values trts = [] branches = [] branchsetids = set() for branching_level in self._ltnode: for branchset in _bsnodes(self.filename, branching_level): if branchset['uncertaintyType'] != 'gmpeModel': raise InvalidLogicTree( '%s: only uncertainties of type "gmpeModel" ' 'are allowed in gmpe logic tree' % self.filename) bsid = branchset['branchSetID'] if bsid in branchsetids: raise InvalidLogicTree( '%s: Duplicated branchSetID %s' % (self.filename, bsid)) else: branchsetids.add(bsid) trt = branchset.attrib.get('applyToTectonicRegionType') if trt: trts.append(trt) # NB: '*' is used in scenario calculations to disable filtering effective = (tectonic_region_types == ['*'] or trt in tectonic_region_types) weights = [] branch_ids = [] for branch in branchset: weight = ImtWeight(branch, self.filename) weights.append(weight) branch_id = branch['branchID'] branch_ids.append(branch_id) uncertainty = to_toml(branch.uncertaintyModel) if isinstance(self.filename, str): # a bit hackish: set the GMPE_DIR equal to the # directory where the gsim_logic_tree file is GMPETable.GMPE_DIR = os.path.dirname(self.filename) try: gsim = valid.gsim(uncertainty) except Exception as exc: raise ValueError( "%s in file %s" % (exc, self.filename)) from exc if (isinstance(gsim, GMPETable) and 'gmpe_table' in gsim.kwargs): self.gmpe_tables.add(gsim.kwargs['gmpe_table']) if gsim in self.values[trt]: raise InvalidLogicTree('%s: duplicated gsim %s' % (self.filename, gsim)) if len(weight.dic) > 1: gsim.weight = weight self.values[trt].append(gsim) bt = BranchTuple( branchset['applyToTectonicRegionType'], branch_id, gsim, weight, effective) branches.append(bt) tot = sum(weights) assert tot.is_one(), '%s in branch %s' % (tot, branch_id) if duplicated(branch_ids): raise InvalidLogicTree( 'There where duplicated branchIDs in %s' % self.filename) if len(trts) > len(set(trts)): raise InvalidLogicTree( '%s: Found duplicated applyToTectonicRegionType=%s' % (self.filename, trts)) branches.sort(key=lambda b: (b.trt, # TODO: add an .idx to each GSIM ? return branches
[docs] def get_gsims(self, trt): """ :param trt: tectonic region type :returns: sorted list of available GSIMs for that trt """ if trt == '*' or trt == b'*': # fake logictree [trt] = self.values return sorted(self.values[trt])
[docs] def sample(self, n, seed): """ :param n: number of samples :param seed: random seed :returns: n Realization objects """ brlists = [sample([b for b in self.branches if b.trt == trt], n, seed + i) for i, trt in enumerate(self.values)] rlzs = [] for i in range(n): weight = 1 lt_path = [] lt_uid = [] value = [] for brlist in brlists: # there is branch list for each TRT branch = brlist[i] lt_path.append( lt_uid.append( if branch.effective else '@') weight *= branch.weight value.append(branch.gsim) rlz = Realization(tuple(value), weight, tuple(lt_path), i, tuple(lt_uid)) rlzs.append(rlz) return rlzs
def __iter__(self): """ Yield :class:`openquake.commonlib.logictree.Realization` instances """ groups = [] # NB: branches are already sorted for trt in self.values: groups.append([b for b in self.branches if b.trt == trt]) # with T tectonic region types there are T groups and T branches for i, branches in enumerate(itertools.product(*groups)): weight = 1 lt_path = [] lt_uid = [] value = [] for trt, branch in zip(self.values, branches): lt_path.append( lt_uid.append( if branch.effective else '@') weight *= branch.weight value.append(branch.gsim) yield Realization(tuple(value), weight, tuple(lt_path), i, tuple(lt_uid)) def __repr__(self): lines = ['%s,%s,%s,w=%s' % (b.trt,, b.gsim, b.weight['weight']) for b in self.branches if b.effective] return '<%s\n%s>' % (self.__class__.__name__, '\n'.join(lines))
[docs]def taxonomy_mapping(filename, taxonomies): """ :param filename: path to the CSV file containing the taxonomy associations :param taxonomies: an array taxonomy string -> taxonomy index :returns: (array, [[(taxonomy, weight), ...], ...]) """ if filename is None: # trivial mapping return (), [[(taxo, 1)] for taxo in taxonomies] dic = {} # taxonomy index -> risk taxonomy array = hdf5.read_csv(filename, {None: hdf5.vstr, 'weight': float}).array arr = add_defaults(array, weight=1.) assert arr.dtype.names == ('taxonomy', 'conversion', 'weight') dic = group_array(arr, 'taxonomy') taxonomies = taxonomies[1:] # strip '?' missing = set(taxonomies) - set(dic) if missing: raise InvalidFile('The taxonomies %s are in the exposure but not in %s' % (missing, filename)) lst = [[("?", 1)]] for idx, taxo in enumerate(taxonomies, 1): recs = dic[taxo] if abs(recs['weight'].sum() - 1.) > pmf.PRECISION: raise InvalidFile('%s: the weights do not sum up to 1 for %s' % (filename, taxo)) lst.append([(rec['conversion'], rec['weight']) for rec in recs]) return arr, lst