Skip to content

logic_tree_base

logic_tree_base

This module contains base classes (some of which are abstract) common to both Source and Ground Motion Model (GMM) logic trees.

BranchSet dataclass

Bases: Generic[BranchType]

A group of branches that comprise their own sub-logic tree. Also known as a fault system logic tree (for source logic trees).

Parameters:

Name Type Description Default
short_name str
''
long_name str
''
branches List[BranchType]

the branches that make up the branch set

list()
Source code in nzshm_model/logic_tree/logic_tree_base.py
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@dataclass
class BranchSet(Generic[BranchType]):
    """
    A group of branches that comprise their own sub-logic tree. Also known as a fault system logic
    tree (for source logic trees).

    Arguments:
        short_name:
        long_name:
        branches: the branches that make up the branch set
    """

    short_name: str = ''
    long_name: str = ''
    branches: List[BranchType] = field(default_factory=list)

    def __post_init__(self):
        helpers._validate_branchset_weights(self)

    def __str__(self) -> str:
        string = f'short_name: {self.short_name}, long_name: {self.long_name}\n'
        string += '======BRANCHES======\n'
        return string + '\n'.join([str(branch) for branch in self])

    def __iter__(self: BranchSetType) -> BranchSetType:
        self.__counter = 0
        return self

    def __next__(self) -> BranchType:
        if self.__counter >= len(self.branches):
            raise StopIteration
        else:
            self.__counter += 1
            return self.branches[self.__counter - 1]

FilteredBranch dataclass

Bases: Branch

A branch type that points back to it's logic tree and branch set. Should never be serialized, only used for filtering

Source code in nzshm_model/logic_tree/logic_tree_base.py
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
@dataclass
class FilteredBranch(Branch):
    """
    A branch type that points back to it's logic tree and branch set. Should never be serialized, only
    used for filtering
    """

    logic_tree: LogicTree = field(default_factory=LogicTree)
    branch_set: BranchSet = field(default_factory=BranchSet)

    def to_branch(self) -> Branch:
        """
        Produce the Branch object (does not point back to it's logic tree and branch set)

        Returns:
            branch: the Branch object
        """
        branch_attributes = {k: v for k, v in self.__dict__.items() if k not in ('branch_set', 'logic_tree')}
        return type(self.branch_set.branches[0])(**branch_attributes)

to_branch()

Produce the Branch object (does not point back to it's logic tree and branch set)

Returns:

Name Type Description
branch Branch

the Branch object

Source code in nzshm_model/logic_tree/logic_tree_base.py
336
337
338
339
340
341
342
343
344
def to_branch(self) -> Branch:
    """
    Produce the Branch object (does not point back to it's logic tree and branch set)

    Returns:
        branch: the Branch object
    """
    branch_attributes = {k: v for k, v in self.__dict__.items() if k not in ('branch_set', 'logic_tree')}
    return type(self.branch_set.branches[0])(**branch_attributes)

LogicTree dataclass

Bases: ABC, Generic[FilteredBranchType]

Logic tree baseclass. Contains information about branch sets and correlations between branches of the branch sets.

The correlations are enforced when forming composite branches (combinations of branches from different branch sets).

Parameters:

Name Type Description Default
title str

title of the logic tree

''
version str

version string

''
branch_sets List[Any]

list of branch sets that make up the logic tree

list()
correlations LogicTreeCorrelations

any correlations between branches of the branch_sets

LogicTreeCorrelations()
Source code in nzshm_model/logic_tree/logic_tree_base.py
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
@dataclass
class LogicTree(ABC, Generic[FilteredBranchType]):
    """
    Logic tree baseclass. Contains information about branch sets and correlations between branches of the branch sets.

    The correlations are enforced when forming composite branches (combinations of branches from different branch sets).

    Arguments:
        title: title of the logic tree
        version: version string
        branch_sets: list of branch sets that make up the logic tree
        correlations: any correlations between branches of the branch_sets

    """

    title: str = ''
    version: str = ''
    branch_sets: List[Any] = field(default_factory=list)
    correlations: LogicTreeCorrelations = field(default_factory=LogicTreeCorrelations)

    def __post_init__(self) -> None:
        # TODO: branch set short_names should be unique (see from_branches())
        helpers._validate_correlation_weights(self)

    def __setattr__(self, __name: str, __value: Any) -> None:
        super().__setattr__(__name, __value)
        if __name == "correlations":
            helpers._validate_correlation_weights(self)

    def __str__(self) -> str:
        string = f'LogicTree type: {type(self)}\n'
        string += f'title: {self.title}, version:{self.version}\n'
        string += '==========BRANCH SETS==========\n\n'
        string += '\n\n'.join([str(bs) for bs in self.branch_sets])
        return string

    def _composite_branches(self) -> Generator[CompositeBranch, None, None]:
        """
        Yields all composite (combined) branches of the branch_sets without applying correlations.

        Returns:
            composite_branches: the CompositeBranches of the combined logic tree BranchSets
        """
        for branches in product(*[branch_set.branches for branch_set in self.branch_sets]):
            yield CompositeBranch(branches=branches)

    @property
    def composite_branches(self) -> Generator[CompositeBranch, None, None]:
        """
        Yields all composite (combined) branches of the branch_sets enforcing correlations.

        Returns:
            composite_branches: the CompositeBranches of the combined logic tree BranchSets
        """
        for composite_branch in self._composite_branches():
            # if the comp_branch contains a branch listed as the 0th element of the correlations, only
            # yeild if the other branches are present
            # weight is automatically calculated by CompositeBranch if not explicitly assigned
            # (as we would with correlations)
            correlation_match = [branch_pri in composite_branch for branch_pri in self.correlations.primary_branches()]
            if any(correlation_match):
                # index of the correlation that matches a branch in _composite_branches()
                i_cor = correlation_match.index(True)
                if not all(br in composite_branch for br in self.correlations[i_cor].all_branches):
                    continue
                weights = [self.correlations[i_cor].weight] + [
                    branch.weight for branch in composite_branch if branch not in self.correlations[i_cor].all_branches
                ]
                composite_branch.weight = reduce(mul, weights, 1.0)
            yield composite_branch

    @classmethod
    def from_json(cls: Type[LogicTreeType], json_path: Union[Path, str]) -> LogicTreeType:
        """
        Create LogicTree object from json file

        See docs/api/logic_tree/source_logic_tree_config_format.md and
        api/logic_tree/gmcm_logic_tree_config_format.md

        Parameters:
            json_path: path to json file

        Returns:
            logic_tree
        """
        with Path(json_path).open() as jsonfile:
            data = json.load(jsonfile)
        return cls.from_dict(data)

    @classmethod
    def from_dict(cls: Type[LogicTreeType], data: Dict) -> LogicTreeType:
        """
        Create LogicTree object from dict.

        See docs/api/logic_tree/source_logic_tree_config_format.md and
        api/logic_tree/gmcm_logic_tree_config_format.md

        Parameters:
            data: dict representation of LogicTree object

        Returns:
            logic_tree
        """
        if not data.get('correlations'):
            logic_tree = cls._from_dict(data)
            # do not need to validate names as that is only necessary if there are correlations
            return logic_tree

        correlations = data.pop('correlations')
        helpers._validate_correlations_format(correlations)
        logic_tree = cls._from_dict(data)
        helpers._validate_names(logic_tree)
        logic_tree = helpers._add_corellations(logic_tree, correlations)

        return logic_tree

    @classmethod
    def _from_dict(cls: Type[LogicTreeType], data: Dict) -> LogicTreeType:
        """
        Create LogicTree object from dict. Input dict must be a direct asdict() serialisation of the LogicTree object

        Parameters:
            data: dict representation of LogicTree object

        Returns:
            logic_tree
        """

        config = dacite.Config(strict=True, cast=[tuple])
        return dacite.from_dict(data_class=cls, data=data, config=config)

    def _to_dict(self) -> Dict[str, Any]:
        """
        Create dict representation of logic tree. This creates an exact dict of the class and is not
        used for serialisation.

        Returns:
            dict:
        """
        return asdict(self)

    def to_dict(self) -> Dict[str, Any]:
        """Create dictionary representation of logic tree used for serialisation.

        Returns:
            logic_tree_dict: dictionary representaion of logic tree
        """

        # do not need to validate names as that is only necessary if there are correlations
        data = self._to_dict()
        if not self.correlations:
            del data["correlations"]
            return data

        helpers._validate_names(self)
        data["correlations"] = helpers._serialise_correlations(self)
        return data

    def to_json(self, file_path: Union[Path, str]) -> None:
        """Serialze logic tree as json file.

        Parameters:
            file_path: path to json file to be written
        """
        file_path = Path(file_path)
        with file_path.open('w') as jsonfile:
            json.dump(self.to_dict(), jsonfile, indent=2)

    def __all_branches__(self) -> Generator[FilteredBranchType, None, None]:
        """
        Yield all branches from all BranchSets, each with a shallow copy of its LogicTree and BranchSet parents
        for use in filtering.

        NB this class is never used for serialising models.
        """

        def get_fields(obj):
            return {
                field.name: copy.deepcopy(getattr(obj, field.name))
                for field in fields(obj)
                if field.name not in ('branches', 'branch_sets')
            }

        lt_fields = get_fields(self)
        for branch_set in self.branch_sets:
            bs_fields = get_fields(branch_set)
            for branch in branch_set.branches:
                bs_lite = type(branch_set)(**bs_fields, branches=[type(branch)()])
                lt_lite = type(self)(**lt_fields)
                # b_fields = get_fields(branch)
                yield branch.filtered_branch(
                    branch_set=bs_lite,
                    logic_tree=lt_lite,
                )

    @classmethod
    def from_branches(cls, branches: Iterator[FilteredBranchType]) -> 'LogicTree':
        """
        Build a complete LogicTree from a iterable of branches.

        We expect that all the branches have come from a single LogicTree.

        Parameters:
            branches: the branches used to build the LogicTree
        """

        def match_branch_set(slt: LogicTree, fb):
            for branch_set in slt.branch_sets:
                if fb.branch_set.short_name == branch_set.short_name:
                    return branch_set

        version = None
        for fb in branches:
            # ensure an slt
            if version is None:
                logic_tree = cls(version=fb.logic_tree.version, title=fb.logic_tree.title)
                version = fb.logic_tree.version
            else:
                assert version == fb.logic_tree.version

            # ensure an branch_set
            bs = match_branch_set(logic_tree, fb)
            if not bs:
                bs = type(fb.branch_set)(short_name=fb.branch_set.short_name, long_name=fb.branch_set.long_name)
                logic_tree.branch_sets.append(bs)
            bs.branches.append(fb.to_branch())
        return logic_tree

    def __iter__(self):
        self.__current_branch = 0
        self.__branch_list = list(self.__all_branches__())
        return self

    def __next__(self) -> FilteredBranchType:
        if self.__current_branch >= len(self.__branch_list):
            raise StopIteration
        else:
            self.__current_branch += 1
            return self.__branch_list[self.__current_branch - 1]

    def psha_adapter(self, provider: Type[PshaAdapterInterface], **kwargs: Optional[Dict]) -> "PshaAdapterInterface":
        """get a PSHA adapter for this instance.

        Arguments:
            provider: the adapter class
            **kwargs: additional arguments required by the provider class

        Returns:
            a PSHA Adapter instance
        """
        return provider(target=self)

composite_branches property

Yields all composite (combined) branches of the branch_sets enforcing correlations.

Returns:

Name Type Description
composite_branches None

the CompositeBranches of the combined logic tree BranchSets

from_branches(branches) classmethod

Build a complete LogicTree from a iterable of branches.

We expect that all the branches have come from a single LogicTree.

Parameters:

Name Type Description Default
branches Iterator[FilteredBranchType]

the branches used to build the LogicTree

required
Source code in nzshm_model/logic_tree/logic_tree_base.py
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
@classmethod
def from_branches(cls, branches: Iterator[FilteredBranchType]) -> 'LogicTree':
    """
    Build a complete LogicTree from a iterable of branches.

    We expect that all the branches have come from a single LogicTree.

    Parameters:
        branches: the branches used to build the LogicTree
    """

    def match_branch_set(slt: LogicTree, fb):
        for branch_set in slt.branch_sets:
            if fb.branch_set.short_name == branch_set.short_name:
                return branch_set

    version = None
    for fb in branches:
        # ensure an slt
        if version is None:
            logic_tree = cls(version=fb.logic_tree.version, title=fb.logic_tree.title)
            version = fb.logic_tree.version
        else:
            assert version == fb.logic_tree.version

        # ensure an branch_set
        bs = match_branch_set(logic_tree, fb)
        if not bs:
            bs = type(fb.branch_set)(short_name=fb.branch_set.short_name, long_name=fb.branch_set.long_name)
            logic_tree.branch_sets.append(bs)
        bs.branches.append(fb.to_branch())
    return logic_tree

from_dict(data) classmethod

Create LogicTree object from dict.

See docs/api/logic_tree/source_logic_tree_config_format.md and api/logic_tree/gmcm_logic_tree_config_format.md

Parameters:

Name Type Description Default
data Dict

dict representation of LogicTree object

required

Returns:

Type Description
LogicTreeType

logic_tree

Source code in nzshm_model/logic_tree/logic_tree_base.py
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@classmethod
def from_dict(cls: Type[LogicTreeType], data: Dict) -> LogicTreeType:
    """
    Create LogicTree object from dict.

    See docs/api/logic_tree/source_logic_tree_config_format.md and
    api/logic_tree/gmcm_logic_tree_config_format.md

    Parameters:
        data: dict representation of LogicTree object

    Returns:
        logic_tree
    """
    if not data.get('correlations'):
        logic_tree = cls._from_dict(data)
        # do not need to validate names as that is only necessary if there are correlations
        return logic_tree

    correlations = data.pop('correlations')
    helpers._validate_correlations_format(correlations)
    logic_tree = cls._from_dict(data)
    helpers._validate_names(logic_tree)
    logic_tree = helpers._add_corellations(logic_tree, correlations)

    return logic_tree

from_json(json_path) classmethod

Create LogicTree object from json file

See docs/api/logic_tree/source_logic_tree_config_format.md and api/logic_tree/gmcm_logic_tree_config_format.md

Parameters:

Name Type Description Default
json_path Union[Path, str]

path to json file

required

Returns:

Type Description
LogicTreeType

logic_tree

Source code in nzshm_model/logic_tree/logic_tree_base.py
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@classmethod
def from_json(cls: Type[LogicTreeType], json_path: Union[Path, str]) -> LogicTreeType:
    """
    Create LogicTree object from json file

    See docs/api/logic_tree/source_logic_tree_config_format.md and
    api/logic_tree/gmcm_logic_tree_config_format.md

    Parameters:
        json_path: path to json file

    Returns:
        logic_tree
    """
    with Path(json_path).open() as jsonfile:
        data = json.load(jsonfile)
    return cls.from_dict(data)

psha_adapter(provider, **kwargs)

get a PSHA adapter for this instance.

Parameters:

Name Type Description Default
provider Type[PshaAdapterInterface]

the adapter class

required
**kwargs Optional[Dict]

additional arguments required by the provider class

{}

Returns:

Type Description
PshaAdapterInterface

a PSHA Adapter instance

Source code in nzshm_model/logic_tree/logic_tree_base.py
313
314
315
316
317
318
319
320
321
322
323
def psha_adapter(self, provider: Type[PshaAdapterInterface], **kwargs: Optional[Dict]) -> "PshaAdapterInterface":
    """get a PSHA adapter for this instance.

    Arguments:
        provider: the adapter class
        **kwargs: additional arguments required by the provider class

    Returns:
        a PSHA Adapter instance
    """
    return provider(target=self)

to_dict()

Create dictionary representation of logic tree used for serialisation.

Returns:

Name Type Description
logic_tree_dict Dict[str, Any]

dictionary representaion of logic tree

Source code in nzshm_model/logic_tree/logic_tree_base.py
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def to_dict(self) -> Dict[str, Any]:
    """Create dictionary representation of logic tree used for serialisation.

    Returns:
        logic_tree_dict: dictionary representaion of logic tree
    """

    # do not need to validate names as that is only necessary if there are correlations
    data = self._to_dict()
    if not self.correlations:
        del data["correlations"]
        return data

    helpers._validate_names(self)
    data["correlations"] = helpers._serialise_correlations(self)
    return data

to_json(file_path)

Serialze logic tree as json file.

Parameters:

Name Type Description Default
file_path Union[Path, str]

path to json file to be written

required
Source code in nzshm_model/logic_tree/logic_tree_base.py
231
232
233
234
235
236
237
238
239
def to_json(self, file_path: Union[Path, str]) -> None:
    """Serialze logic tree as json file.

    Parameters:
        file_path: path to json file to be written
    """
    file_path = Path(file_path)
    with file_path.open('w') as jsonfile:
        json.dump(self.to_dict(), jsonfile, indent=2)