Skip to content

versioning

versioning

This module provides tools for managing resource versions.

Classes:

Name Description
VersionManager

manage the reading/writing of versions.

VERSION_LIST_FILENAME = 'version_list.json' module-attribute

VersionManager

A class to manage the reading/writing of versions.

Source code in nzssdt_2023/versioning/versioning.py 
 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
 71
 72
 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
class VersionManager:
    """A class to manage the reading/writing of versions."""

    def __init__(self, resource_folder: Optional[str] = None):
        self._resource_folder = resource_folder or RESOURCES_FOLDER
        self._version_list_path = Path(self._resource_folder, VERSION_LIST_FILENAME)

    def read_version_list(self) -> Dict[str, VersionInfo]:
        """return the version list as a dict.

        Returns:
            version_dict: a dictionary of version info instances.
        """
        if not self._version_list_path.exists():
            raise RuntimeError(
                f"the version_list file {self._version_list_path} was not found."
            )
        version_list = json.load(open(self._version_list_path))
        version_dict = {}
        seen = []
        for vi in version_list:
            obj = dacite.from_dict(data_class=VersionInfo, data=vi)
            assert obj.version_id not in seen, "duplicate version id"
            version_dict[obj.version_id] = obj
            seen.append(obj.version_id)
        return version_dict

    def write_version_list(self, new_list: Iterable[VersionInfo]):
        """write the version list.

        Args:
            new_list: the list data to write to disk.
        """
        with open(self._version_list_path, "w") as fout:
            json.dump([dataclasses.asdict(vi) for vi in new_list], fout, indent=2)

    def get(self, version_id: str) -> Union[VersionInfo, None]:
        """get a version, given a valid id.

        Args:
            version_id: the version id string.

        Returns:
            version_info: the version info instance
        """
        versions = self.read_version_list()
        return versions.get(version_id)

    def add(self, version_info: VersionInfo):
        """add a version instance.

        Args:
            version_info: the version_info instance.

        Raises:
            KeyError: if the version_id already exists.
        """
        versions = self.read_version_list()
        if versions.get(version_info.version_id):
            raise (KeyError("Item already exists"))
        versions[version_info.version_id] = version_info
        self.write_version_list(list(versions.values()))

    def update(self, version_info: VersionInfo):
        """update a version instance.

        Args:
            version_info: the modified version instance.

        Raises:
            KeyError: if the version_id was not found
        """
        versions = self.read_version_list()
        current = versions.get(version_info.version_id, None)
        if not current:
            raise (KeyError)
        versions[version_info.version_id] = version_info
        self.write_version_list(versions.values())

    def remove(self, version_id: str) -> VersionInfo:
        """remove a version by version_id.

        Args:
            version_id: the version id string.

        Returns:
            version_info: the removed version info instance
        """
        versions = self.read_version_list()
        vi = versions.pop(version_id)
        self.write_version_list(versions.values())
        return vi

__init__(resource_folder: Optional[str] = None)

Source code in nzssdt_2023/versioning/versioning.py 
42
43
44
def __init__(self, resource_folder: Optional[str] = None):
    self._resource_folder = resource_folder or RESOURCES_FOLDER
    self._version_list_path = Path(self._resource_folder, VERSION_LIST_FILENAME)

add(version_info: VersionInfo)

add a version instance.

Parameters:

Name Type Description Default
version_info VersionInfo

the version_info instance.

 required

Raises:

Type Description
KeyError

if the version_id already exists.
Source code in nzssdt_2023/versioning/versioning.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def add(self, version_info: VersionInfo):
    """add a version instance.

    Args:
        version_info: the version_info instance.

    Raises:
        KeyError: if the version_id already exists.
    """
    versions = self.read_version_list()
    if versions.get(version_info.version_id):
        raise (KeyError("Item already exists"))
    versions[version_info.version_id] = version_info
    self.write_version_list(list(versions.values()))

get(version_id: str) -> Union[VersionInfo, None]

get a version, given a valid id.

Parameters:

Name Type Description Default
version_id str

the version id string.

 required

Returns:

Name Type Description
version_info Union[VersionInfo, None]

the version info instance
Source code in nzssdt_2023/versioning/versioning.py 
75
76
77
78
79
80
81
82
83
84
85
def get(self, version_id: str) -> Union[VersionInfo, None]:
    """get a version, given a valid id.

    Args:
        version_id: the version id string.

    Returns:
        version_info: the version info instance
    """
    versions = self.read_version_list()
    return versions.get(version_id)

read_version_list() -> Dict[str, VersionInfo]

return the version list as a dict.

Returns:

Name Type Description
version_dict Dict[str, VersionInfo]

a dictionary of version info instances.
Source code in nzssdt_2023/versioning/versioning.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def read_version_list(self) -> Dict[str, VersionInfo]:
    """return the version list as a dict.

    Returns:
        version_dict: a dictionary of version info instances.
    """
    if not self._version_list_path.exists():
        raise RuntimeError(
            f"the version_list file {self._version_list_path} was not found."
        )
    version_list = json.load(open(self._version_list_path))
    version_dict = {}
    seen = []
    for vi in version_list:
        obj = dacite.from_dict(data_class=VersionInfo, data=vi)
        assert obj.version_id not in seen, "duplicate version id"
        version_dict[obj.version_id] = obj
        seen.append(obj.version_id)
    return version_dict

remove(version_id: str) -> VersionInfo

remove a version by version_id.

Parameters:

Name Type Description Default
version_id str

the version id string.

 required

Returns:

Name Type Description
version_info VersionInfo

the removed version info instance
Source code in nzssdt_2023/versioning/versioning.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
def remove(self, version_id: str) -> VersionInfo:
    """remove a version by version_id.

    Args:
        version_id: the version id string.

    Returns:
        version_info: the removed version info instance
    """
    versions = self.read_version_list()
    vi = versions.pop(version_id)
    self.write_version_list(versions.values())
    return vi

update(version_info: VersionInfo)

update a version instance.

Parameters:

Name Type Description Default
version_info VersionInfo

the modified version instance.

 required

Raises:

Type Description
KeyError

if the version_id was not found
Source code in nzssdt_2023/versioning/versioning.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def update(self, version_info: VersionInfo):
    """update a version instance.

    Args:
        version_info: the modified version instance.

    Raises:
        KeyError: if the version_id was not found
    """
    versions = self.read_version_list()
    current = versions.get(version_info.version_id, None)
    if not current:
        raise (KeyError)
    versions[version_info.version_id] = version_info
    self.write_version_list(versions.values())

write_version_list(new_list: Iterable[VersionInfo])

write the version list.

Parameters:

Name Type Description Default
new_list Iterable[VersionInfo]

the list data to write to disk.

 required
Source code in nzssdt_2023/versioning/versioning.py 
66
67
68
69
70
71
72
73
def write_version_list(self, new_list: Iterable[VersionInfo]):
    """write the version list.

    Args:
        new_list: the list data to write to disk.
    """
    with open(self._version_list_path, "w") as fout:
        json.dump([dataclasses.asdict(vi) for vi in new_list], fout, indent=2)

ensure_resource_folders(version_id: str, exist_ok: bool = False)

Source code in nzssdt_2023/versioning/versioning.py 
23
24
25
26
27
28
29
def ensure_resource_folders(version_id: str, exist_ok: bool = False):
    version_folder = Path(RESOURCES_FOLDER).parent / "resources" / f"v{version_id}"
    pipeline_folder = (
        Path(RESOURCES_FOLDER).parent / "resources" / "pipeline" / f"v{version_id}"
    )
    version_folder.mkdir(exist_ok=exist_ok)
    pipeline_folder.mkdir(exist_ok=exist_ok)

standard_output_filename(version: Union[str, VersionInfo])

Source code in nzssdt_2023/versioning/versioning.py 
32
33
34
35
36
def standard_output_filename(version: Union[str, "VersionInfo"]):
    # print(type(version))
    if isinstance(version, VersionInfo):
        version = version.version_id
    return f"nzssdt_2023_v{version}.json.zip"

dataclass

Dataclasses defining structures for versioning.

ConvertedFile dataclass

A dataclass defining a converted file.

NB not used from v2 on

Parameters:

Name Type Description Default
input_filepath str

path to the original file.

 required
output_filepath str

path to the output file.

 required
Source code in nzssdt_2023/versioning/dataclass.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@dataclass(frozen=True)
class ConvertedFile:
    """
    A dataclass defining a converted file.

    NB not used from v2 on

    Args:
        input_filepath: path to the original file.
        output_filepath: path to the output file.
    """

    input_filepath: str
    output_filepath: str

input_filepath: str instance-attribute

output_filepath: str instance-attribute

__init__(input_filepath: str, output_filepath: str) -> None

IncludedFile dataclass

A dataclass defining a resource file.

Parameters:

Name Type Description Default
filepath str

path to the file.

 required
Source code in nzssdt_2023/versioning/dataclass.py 
31
32
33
34
35
36
37
38
39
40
@dataclass(frozen=True)
class IncludedFile:
    """
    A dataclass defining a resource file.

    Args:
        filepath: path to the file.
    """

    filepath: str

filepath: str instance-attribute

__init__(filepath: str) -> None

VersionInfo dataclass

A dataclass defining the attributes of a NZSDDT version.

Parameters:

Name Type Description Default
version_id str

a unique version number.

 required
nzshm_model_version str

the NSHM model version string.

 required
description Optional[str]

a versions description.

 None
conversions List[ConvertedFile]

a list of files converted (from AH to versioned) TODO: not used in v2.

 list()
manifest List[IncludedFile]

the data files used for reporting

 list()
reports List[IncludedFile]

the reports files csv and PDF

 list()
nzshm_common_lib_version str

the version of the nzshm_common library used to produce this version.

 nzshm_common.__version__
Source code in nzssdt_2023/versioning/dataclass.py 
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@dataclass(frozen=True)
class VersionInfo:
    """
    A dataclass defining the attributes of a NZSDDT version.

    Args:
        version_id: a unique version number.
        nzshm_model_version: the NSHM model version string.
        description: a versions description.
        conversions: a list of files converted (from AH to versioned) TODO: not used in v2.
        manifest: the data files used for reporting
        reports: the reports files csv and PDF
        nzshm_common_lib_version: the version of the nzshm_common library used to produce this version.
    """

    version_id: str = field(hash=True)
    nzshm_model_version: str  # nzshm_model.CURRENT_VERSION  # default to latest
    description: Optional[str] = None
    conversions: List[ConvertedFile] = field(default_factory=list)  # not used after v1
    manifest: List[IncludedFile] = field(default_factory=list)
    reports: List[IncludedFile] = field(default_factory=list)
    nzshm_common_lib_version: str = nzshm_common.__version__

    def __str__(self):
        return f"version: {self.version_id}, model: {self.nzshm_model_version}, description: `{self.description}`"

    def resource_path(self, resource_folder: Optional[str] = None) -> Path:
        rf = resource_folder or RESOURCES_FOLDER
        return Path(rf) / f"v{self.version_id}"

    def reports_path(self, reports_folder: Optional[str] = None) -> Path:
        rf = reports_folder or REPORTS_FOLDER
        return Path(rf) / f"v{self.version_id}"

    def collect_manifest(self):
        # update manifest
        resources = self.resource_path()
        for file in resources.iterdir():
            self.manifest.append(
                IncludedFile(str(file.relative_to(resources.parent.parent)))
            )

        reports = self.reports_path()
        for file in reports.iterdir():
            self.reports.append(
                IncludedFile(str(file.relative_to(reports.parent.parent)))
            )

conversions: List[ConvertedFile] = field(default_factory=list) class-attribute instance-attribute

description: Optional[str] = None class-attribute instance-attribute

manifest: List[IncludedFile] = field(default_factory=list) class-attribute instance-attribute

nzshm_common_lib_version: str = nzshm_common.__version__ class-attribute instance-attribute

nzshm_model_version: str instance-attribute

reports: List[IncludedFile] = field(default_factory=list) class-attribute instance-attribute

version_id: str = field(hash=True) class-attribute instance-attribute

__init__(version_id: str, nzshm_model_version: str, description: Optional[str] = None, conversions: List[ConvertedFile] = list(), manifest: List[IncludedFile] = list(), reports: List[IncludedFile] = list(), nzshm_common_lib_version: str = nzshm_common.__version__) -> None

__str__()

Source code in nzssdt_2023/versioning/dataclass.py 
66
67
def __str__(self):
    return f"version: {self.version_id}, model: {self.nzshm_model_version}, description: `{self.description}`"

collect_manifest()

Source code in nzssdt_2023/versioning/dataclass.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
def collect_manifest(self):
    # update manifest
    resources = self.resource_path()
    for file in resources.iterdir():
        self.manifest.append(
            IncludedFile(str(file.relative_to(resources.parent.parent)))
        )

    reports = self.reports_path()
    for file in reports.iterdir():
        self.reports.append(
            IncludedFile(str(file.relative_to(reports.parent.parent)))
        )

reports_path(reports_folder: Optional[str] = None) -> Path

Source code in nzssdt_2023/versioning/dataclass.py 
73
74
75
def reports_path(self, reports_folder: Optional[str] = None) -> Path:
    rf = reports_folder or REPORTS_FOLDER
    return Path(rf) / f"v{self.version_id}"

resource_path(resource_folder: Optional[str] = None) -> Path

Source code in nzssdt_2023/versioning/dataclass.py 
69
70
71
def resource_path(self, resource_folder: Optional[str] = None) -> Path:
    rf = resource_folder or RESOURCES_FOLDER
    return Path(rf) / f"v{self.version_id}"