Skip to content

branch_registry

branch_registry

A registry for logic_tree_branches (sources and GMMs) that are uniquely identified by a shake_256 hash.

Data is stored in the resources folder in CSV form. New branched may be added by developers to support new source or logic tree components.

Examples:

>>> from nzshm_model import branch_registry
>>> registry = branch_registry.Registry()
>>> entry = registry.source_registry.get_by_hash("af9ec2b004d7")
... BranchRegistryEntry(... )

Functions:

Name Description
identity_digest

get a standard hash_digest from an identity string

BranchRegistry

Storage Manager for BranchRegistryEntry objects

Source code in nzshm_model/branch_registry.py 
 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
class BranchRegistry:
    """Storage Manager for BranchRegistryEntry objects"""

    def __init__(self):
        self._branches_by_hash = dict()
        self._branches_by_identity = dict()
        self._branches_by_extra = dict()

    def _load_row(self, row):
        entry = BranchRegistryEntry(**row)
        assert entry.hash_digest not in self._branches_by_hash
        assert entry.identity not in self._branches_by_identity

        self._branches_by_hash[entry.hash_digest] = entry
        self._branches_by_identity[entry.identity] = entry
        if entry.extra:
            self._branches_by_extra[entry.extra] = entry

    def load(self, registry_file: IO[Any]) -> 'BranchRegistry':
        """Load the entries contained in a CSV file.

        Arguments:
            registry_file: file-like object with expected CSV header file

        Returns:
            the populated BranchRegistry
        """
        registry_file.seek(0)
        reader = csv.DictReader(registry_file, fieldnames=HEADERS)
        headers = list(next(reader).values())
        assert HEADERS == headers
        for row in reader:
            self._load_row(row)
        return self

    def save(self, registry_file: IO[Any]) -> None:
        """Save the registry entries in CSV format.

        Arguments:
            registry_file: file-like object to write
        """
        csv_writer = csv.DictWriter(registry_file, fieldnames=HEADERS)
        csv_writer.writeheader()
        for row in self._branches_by_hash.values():
            csv_writer.writerow(asdict(row))

    def add(self, entry: BranchRegistryEntry) -> None:
        """add a new registry entry.

        Arguments:
            entry: a BranchRegistryEntry object.
        """
        if entry.hash_digest in self._branches_by_hash:
            assert entry.identity in self._branches_by_identity
        else:
            assert entry.identity not in self._branches_by_identity

        self._branches_by_hash[entry.hash_digest] = entry
        self._branches_by_identity[entry.identity] = entry

    def get_by_hash(self, hash_digest: str) -> BranchRegistryEntry:
        """Get a registry entry by hash_digest.

        Arguments:
            hash_digest: the hash digest string.
        """
        return self._branches_by_hash[hash_digest]

    def get_by_identity(self, identity: str) -> BranchRegistryEntry:
        """Get a registry entry by identity string.

        Arguments:
            identity: the identity string.
        """
        return self._branches_by_identity[identity]

    def get_by_extra(self, extra: str) -> Union[BranchRegistryEntry, None]:
        """Get a registry entry by the extra string.

        Notes:
         - this may return None.
         - only used a workaroound becuase some post NSHM hazard jobs used the extra value
           instead of the identity string in the HDF% ( e.g. `T3BlbnF1YWtlSGF6YXJkVGFzazo2OTMxODkz` )

        Arguments:
            extra: the extra string.
        """
        return self._branches_by_extra.get(extra)

    def __len__(self):
        return len(self._branches_by_hash.keys())

add(entry)

add a new registry entry.

Parameters:

Name Type Description Default
entry BranchRegistryEntry

a BranchRegistryEntry object.

 required
Source code in nzshm_model/branch_registry.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
def add(self, entry: BranchRegistryEntry) -> None:
    """add a new registry entry.

    Arguments:
        entry: a BranchRegistryEntry object.
    """
    if entry.hash_digest in self._branches_by_hash:
        assert entry.identity in self._branches_by_identity
    else:
        assert entry.identity not in self._branches_by_identity

    self._branches_by_hash[entry.hash_digest] = entry
    self._branches_by_identity[entry.identity] = entry

get_by_extra(extra)

Get a registry entry by the extra string.

Notes
  • this may return None.
  • only used a workaroound becuase some post NSHM hazard jobs used the extra value instead of the identity string in the HDF% ( e.g. T3BlbnF1YWtlSGF6YXJkVGFzazo2OTMxODkz )

Parameters:

Name Type Description Default
extra str

the extra string.

 required
Source code in nzshm_model/branch_registry.py 
163
164
165
166
167
168
169
170
171
172
173
174
def get_by_extra(self, extra: str) -> Union[BranchRegistryEntry, None]:
    """Get a registry entry by the extra string.

    Notes:
     - this may return None.
     - only used a workaroound becuase some post NSHM hazard jobs used the extra value
       instead of the identity string in the HDF% ( e.g. `T3BlbnF1YWtlSGF6YXJkVGFzazo2OTMxODkz` )

    Arguments:
        extra: the extra string.
    """
    return self._branches_by_extra.get(extra)

get_by_hash(hash_digest)

Get a registry entry by hash_digest.

Parameters:

Name Type Description Default
hash_digest str

the hash digest string.

 required
Source code in nzshm_model/branch_registry.py 
147
148
149
150
151
152
153
def get_by_hash(self, hash_digest: str) -> BranchRegistryEntry:
    """Get a registry entry by hash_digest.

    Arguments:
        hash_digest: the hash digest string.
    """
    return self._branches_by_hash[hash_digest]

get_by_identity(identity)

Get a registry entry by identity string.

Parameters:

Name Type Description Default
identity str

the identity string.

 required
Source code in nzshm_model/branch_registry.py 
155
156
157
158
159
160
161
def get_by_identity(self, identity: str) -> BranchRegistryEntry:
    """Get a registry entry by identity string.

    Arguments:
        identity: the identity string.
    """
    return self._branches_by_identity[identity]

load(registry_file)

Load the entries contained in a CSV file.

Parameters:

Name Type Description Default
registry_file IO[Any]

file-like object with expected CSV header file

 required

Returns:

Type Description
BranchRegistry

the populated BranchRegistry
Source code in nzshm_model/branch_registry.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def load(self, registry_file: IO[Any]) -> 'BranchRegistry':
    """Load the entries contained in a CSV file.

    Arguments:
        registry_file: file-like object with expected CSV header file

    Returns:
        the populated BranchRegistry
    """
    registry_file.seek(0)
    reader = csv.DictReader(registry_file, fieldnames=HEADERS)
    headers = list(next(reader).values())
    assert HEADERS == headers
    for row in reader:
        self._load_row(row)
    return self

save(registry_file)

Save the registry entries in CSV format.

Parameters:

Name Type Description Default
registry_file IO[Any]

file-like object to write

 required
Source code in nzshm_model/branch_registry.py 
122
123
124
125
126
127
128
129
130
131
def save(self, registry_file: IO[Any]) -> None:
    """Save the registry entries in CSV format.

    Arguments:
        registry_file: file-like object to write
    """
    csv_writer = csv.DictWriter(registry_file, fieldnames=HEADERS)
    csv_writer.writeheader()
    for row in self._branches_by_hash.values():
        csv_writer.writerow(asdict(row))

BranchRegistryEntry dataclass

A standard registry entry

Attributes:

Name Type Description
identity str

the unique identity string (see Branch.registry_identity)
hash_digest Optional[str]

the shake_256 hexdigest of the identity.
extra Optional[str]

more information about the entry.
Source code in nzshm_model/branch_registry.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@dataclass
class BranchRegistryEntry:
    """A standard registry entry

    Attributes:
        identity: the unique identity string (see Branch.registry_identity)
        hash_digest: the shake_256 hexdigest of the identity.
        extra: more information about the entry.
    """

    identity: str
    hash_digest: Optional[str] = None
    extra: Optional[str] = None

    def __post_init__(self):
        if self.hash_digest:
            if not self.hash_digest == identity_digest(self.identity):
                raise ValueError(f'Incorrect hash_digest "{self.hash_digest}"" for "{self.identity}"')
        else:
            self.hash_digest = identity_digest(self.identity)

Registry

A container for the published registries.

Attributes:

Name Type Description
gmm_registry BranchRegistry

the model sources registry.
source_registry BranchRegistry

the model gmms registry.
Source code in nzshm_model/branch_registry.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class Registry:
    """A container for the published registries.

    Attributes:
        gmm_registry: the model sources registry.
        source_registry: the model gmms registry.
    """

    _gmms: Optional['BranchRegistry'] = None
    _sources: Optional['BranchRegistry'] = None

    @property
    def gmm_registry(self) -> 'BranchRegistry':
        if not self._gmms:
            self._gmms = BranchRegistry().load(GMM_REGISTRY_CSV.open('r'))
        return self._gmms

    @property
    def source_registry(self) -> 'BranchRegistry':
        if not self._sources:
            self._sources = BranchRegistry().load(SOURCE_REGISTRY_CSV.open('r'))
        return self._sources

identity_digest(identity)

get a standard hash_digest from an identity string

Parameters:

Name Type Description Default
identity str

a unique identity string

 required

Returns:

Type Description
str

a shake_256 hash digest
Source code in nzshm_model/branch_registry.py 
29
30
31
32
33
34
35
36
37
38
def identity_digest(identity: str) -> str:
    """get a standard hash_digest from an identity string

    Arguments:
        identity: a unique identity string

    Returns:
        a shake_256 hash digest
    """
    return hashlib.shake_256(identity.encode()).hexdigest(6)