Skip to content

provenance

lacuna.core.provenance

Provenance tracking utilities for reproducibility.

Functions for recording and managing transformation history.

TransformationRecord dataclass

Record of a spatial transformation operation.

Attributes: source_space: Source coordinate space identifier (e.g., 'MNI152NLin6Asym') source_resolution: Source resolution in mm target_space: Target coordinate space identifier target_resolution: Target resolution in mm method: Transformation method (e.g., 'nitransforms') interpolation: Interpolation method used (e.g., 'linear', 'nearest') timestamp: ISO 8601 timestamp of transformation rationale: Optional explanation for transformation strategy choice transform_file: Optional path/identifier of transform file used

Source code in src/lacuna/core/provenance.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@dataclass
class TransformationRecord:
    """Record of a spatial transformation operation.

    Attributes:
        source_space: Source coordinate space identifier (e.g., 'MNI152NLin6Asym')
        source_resolution: Source resolution in mm
        target_space: Target coordinate space identifier
        target_resolution: Target resolution in mm
        method: Transformation method (e.g., 'nitransforms')
        interpolation: Interpolation method used (e.g., 'linear', 'nearest')
        timestamp: ISO 8601 timestamp of transformation
        rationale: Optional explanation for transformation strategy choice
        transform_file: Optional path/identifier of transform file used
    """

    source_space: str
    source_resolution: float
    target_space: str
    target_resolution: float
    method: str
    interpolation: str
    timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    rationale: str | None = None
    transform_file: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for provenance tracking."""
        result = {
            "type": "spatial_transformation",
            "source_space": self.source_space,
            "source_resolution": self.source_resolution,
            "target_space": self.target_space,
            "target_resolution": self.target_resolution,
            "method": self.method,
            "interpolation": self.interpolation,
            "timestamp": self.timestamp,
        }
        if self.rationale is not None:
            result["rationale"] = self.rationale
        if self.transform_file is not None:
            result["transform_file"] = self.transform_file
        return result

to_dict()

Convert to dictionary for provenance tracking.

Source code in src/lacuna/core/provenance.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary for provenance tracking."""
    result = {
        "type": "spatial_transformation",
        "source_space": self.source_space,
        "source_resolution": self.source_resolution,
        "target_space": self.target_space,
        "target_resolution": self.target_resolution,
        "method": self.method,
        "interpolation": self.interpolation,
        "timestamp": self.timestamp,
    }
    if self.rationale is not None:
        result["rationale"] = self.rationale
    if self.transform_file is not None:
        result["transform_file"] = self.transform_file
    return result

create_provenance_record(function, parameters, version, output_space=None)

Create a provenance record for a transformation.

Parameters:

Name Type Description Default
function str

Fully qualified function name (e.g., 'lacuna.analysis.RegionalDamage').

 required
parameters dict

Function parameters (must be JSON-serializable).

 required
version str

Package version at time of execution.

 required
output_space str

Resulting coordinate space (if spatial operation).

 None

Returns:

Type Description
dict

Provenance record with function, parameters, timestamp, version.

Raises:

Type Description
ProvenanceError

If parameters are not JSON-serializable.

Examples:

>>> record = create_provenance_record(
...     function="lacuna.analysis.RegionalDamage",
...     parameters={"parcel_names": ["schaefer2018parcels100networks7"]},
...     version="0.1.0",
... )
>>> record['function']
'lacuna.analysis.RegionalDamage'
Source code in src/lacuna/core/provenance.py 
 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
def create_provenance_record(
    function: str,
    parameters: dict[str, Any],
    version: str,
    output_space: str | None = None,
) -> dict[str, Any]:
    """
    Create a provenance record for a transformation.

    Parameters
    ----------
    function : str
        Fully qualified function name (e.g., 'lacuna.analysis.RegionalDamage').
    parameters : dict
        Function parameters (must be JSON-serializable).
    version : str
        Package version at time of execution.
    output_space : str, optional
        Resulting coordinate space (if spatial operation).

    Returns
    -------
    dict
        Provenance record with function, parameters, timestamp, version.

    Raises
    ------
    ProvenanceError
        If parameters are not JSON-serializable.

    Examples
    --------
    >>> record = create_provenance_record(
    ...     function="lacuna.analysis.RegionalDamage",
    ...     parameters={"parcel_names": ["schaefer2018parcels100networks7"]},
    ...     version="0.1.0",
    ... )
    >>> record['function']
    'lacuna.analysis.RegionalDamage'
    """
    # Validate parameters are serializable
    try:
        import json

        json.dumps(parameters)
    except (TypeError, ValueError) as e:
        raise ProvenanceError(f"Parameters must be JSON-serializable, got error: {e}") from e

    # Create record
    record = {
        "function": function,
        "parameters": parameters,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "version": version,
    }

    if output_space is not None:
        record["output_space"] = output_space

    return record

merge_provenance(base_provenance, new_provenance)

Merge two provenance lists.

Parameters:

Name Type Description Default
base_provenance list

Base provenance history.

 required
new_provenance list

New provenance to append.

 required

Returns:

Type Description
list

Merged provenance list (ordered chronologically).
Source code in src/lacuna/core/provenance.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def merge_provenance(base_provenance: list, new_provenance: list) -> list:
    """
    Merge two provenance lists.

    Parameters
    ----------
    base_provenance : list
        Base provenance history.
    new_provenance : list
        New provenance to append.

    Returns
    -------
    list
        Merged provenance list (ordered chronologically).
    """
    merged = list(base_provenance)  # Copy to avoid mutation
    merged.extend(new_provenance)
    return merged

validate_provenance_record(record)

Validate a provenance record structure.

Parameters:

Name Type Description Default
record dict

Provenance record to validate.

 required

Raises:

Type Description
ProvenanceError

If record structure is invalid.
Source code in src/lacuna/core/provenance.py 
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
def validate_provenance_record(record: dict[str, Any]) -> None:
    """
    Validate a provenance record structure.

    Parameters
    ----------
    record : dict
        Provenance record to validate.

    Raises
    ------
    ProvenanceError
        If record structure is invalid.
    """
    required_fields = ["function", "parameters", "timestamp", "version"]

    for field_name in required_fields:
        if field_name not in record:
            raise ProvenanceError(f"Provenance record missing required field: {field_name}")

    # Validate timestamp format
    try:
        datetime.fromisoformat(record["timestamp"])
    except (ValueError, TypeError) as e:
        raise ProvenanceError(f"Invalid timestamp format: {record['timestamp']}") from e

    # Validate parameters are dict
    if not isinstance(record["parameters"], dict):
        raise ProvenanceError(f"Parameters must be dict, got {type(record['parameters'])}")