Skip to content

Connectivity#

Wrapper for UGRID connectivity arrays that handles start_index normalization (always 0-indexed internally) and _FillValue masking for mixed-element meshes.

pyramids.netcdf.ugrid.Connectivity dataclass #

Wrapper for UGRID connectivity arrays.

Handles start_index normalization (always 0-indexed internally) and fill_value masking for mixed-element meshes. Connectivity arrays map between mesh elements (e.g., face-to-node, edge-to-node).

Attributes:

Name Type Description
data ndarray

Index array, always 0-indexed internally. Shape is (n_elements, max_nodes_per_element).
fill_value int

Value used to pad rows for elements with fewer nodes than max_nodes_per_element. Default: -1.
cf_role str

UGRID cf_role string (e.g., "face_node_connectivity").
original_start_index int

Original start_index from the file (0 or 1).
Source code in src/pyramids/netcdf/ugrid/connectivity.py 
 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
 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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@dataclass
class Connectivity:
    """Wrapper for UGRID connectivity arrays.

    Handles start_index normalization (always 0-indexed internally)
    and fill_value masking for mixed-element meshes. Connectivity
    arrays map between mesh elements (e.g., face-to-node, edge-to-node).

    Attributes:
        data: Index array, always 0-indexed internally. Shape is
            (n_elements, max_nodes_per_element).
        fill_value: Value used to pad rows for elements with fewer
            nodes than max_nodes_per_element. Default: -1.
        cf_role: UGRID cf_role string (e.g., "face_node_connectivity").
        original_start_index: Original start_index from the file (0 or 1).
    """

    data: np.ndarray
    fill_value: int
    cf_role: str
    original_start_index: int

    @classmethod
    def from_gdal_array(cls, md_arr: gdal.MDArray, cf_role: str) -> Connectivity:
        """Read connectivity from a GDAL MDArray, normalizing start_index.

        Reads the raw index array, subtracts start_index if it is 1,
        and replaces the original fill value with -1 for internal use.

        Args:
            md_arr: GDAL MDArray containing the connectivity data.
            cf_role: The cf_role attribute value for this connectivity.

        Returns:
            Connectivity instance with 0-indexed data.
        """
        attrs = _read_attributes(md_arr)
        start_index = int(attrs.get("start_index", 0))
        raw_fill = attrs.get("_FillValue")
        if raw_fill is not None:
            raw_fill = int(raw_fill)
        else:
            raw_fill = -999

        raw_data = md_arr.ReadAsArray()
        if raw_data is None:
            raise ValueError(
                f"Cannot read connectivity array for cf_role='{cf_role}'."
            )
        data = raw_data.copy().astype(np.intp)

        mask = data == raw_fill

        if start_index != 0:
            data[~mask] -= start_index

        data[mask] = -1

        result = cls(
            data=data,
            fill_value=-1,
            cf_role=cf_role,
            original_start_index=start_index,
        )
        return result

    @property
    def n_elements(self) -> int:
        """Number of elements (rows in the connectivity array)."""
        result = self.data.shape[0]
        return result

    @property
    def max_nodes_per_element(self) -> int:
        """Maximum number of nodes per element (columns)."""
        result = self.data.shape[1] if self.data.ndim > 1 else 1
        return result

    def get_element(self, idx: int) -> np.ndarray:
        """Return valid node indices for a single element.

        Excludes fill values, returning only the actual node indices
        that define this element.

        Args:
            idx: Element index (row in the connectivity array).

        Returns:
            1D array of valid node indices for the element.
        """
        row = self.data[idx]
        if self.data.ndim == 1:
            result = np.atleast_1d(row)
        else:
            result = row[row != self.fill_value]
        return result

    def nodes_per_element(self) -> np.ndarray:
        """Return the number of valid nodes per element.

        Returns:
            1D integer array of length n_elements with the count
            of valid (non-fill) nodes for each element.
        """
        if self.data.ndim == 1:
            result = np.ones(self.n_elements, dtype=np.intp)
        else:
            result = np.sum(self.data != self.fill_value, axis=1).astype(np.intp)
        return result

    def is_triangular(self) -> bool:
        """Check if all elements have exactly 3 nodes.

        Returns:
            True if every element has exactly 3 valid nodes.
        """
        counts = self.nodes_per_element()
        result = bool(np.all(counts == 3))
        return result

    def as_masked(self) -> np.ma.MaskedArray:
        """Return a masked array with fill values masked.

        Returns:
            MaskedArray where positions with fill_value are masked.
        """
        result = np.ma.MaskedArray(self.data, mask=(self.data == self.fill_value))
        return result

n_elements property #

Number of elements (rows in the connectivity array).

max_nodes_per_element property #

Maximum number of nodes per element (columns).

from_gdal_array(md_arr, cf_role) classmethod #

Read connectivity from a GDAL MDArray, normalizing start_index.

Reads the raw index array, subtracts start_index if it is 1, and replaces the original fill value with -1 for internal use.

Parameters:

Name Type Description Default
md_arr MDArray

GDAL MDArray containing the connectivity data.

 required
cf_role str

The cf_role attribute value for this connectivity.

 required

Returns:

Type Description
Connectivity

Connectivity instance with 0-indexed data.
Source code in src/pyramids/netcdf/ugrid/connectivity.py 
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
@classmethod
def from_gdal_array(cls, md_arr: gdal.MDArray, cf_role: str) -> Connectivity:
    """Read connectivity from a GDAL MDArray, normalizing start_index.

    Reads the raw index array, subtracts start_index if it is 1,
    and replaces the original fill value with -1 for internal use.

    Args:
        md_arr: GDAL MDArray containing the connectivity data.
        cf_role: The cf_role attribute value for this connectivity.

    Returns:
        Connectivity instance with 0-indexed data.
    """
    attrs = _read_attributes(md_arr)
    start_index = int(attrs.get("start_index", 0))
    raw_fill = attrs.get("_FillValue")
    if raw_fill is not None:
        raw_fill = int(raw_fill)
    else:
        raw_fill = -999

    raw_data = md_arr.ReadAsArray()
    if raw_data is None:
        raise ValueError(
            f"Cannot read connectivity array for cf_role='{cf_role}'."
        )
    data = raw_data.copy().astype(np.intp)

    mask = data == raw_fill

    if start_index != 0:
        data[~mask] -= start_index

    data[mask] = -1

    result = cls(
        data=data,
        fill_value=-1,
        cf_role=cf_role,
        original_start_index=start_index,
    )
    return result

get_element(idx) #

Return valid node indices for a single element.

Excludes fill values, returning only the actual node indices that define this element.

Parameters:

Name Type Description Default
idx int

Element index (row in the connectivity array).

 required

Returns:

Type Description
ndarray

1D array of valid node indices for the element.
Source code in src/pyramids/netcdf/ugrid/connectivity.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def get_element(self, idx: int) -> np.ndarray:
    """Return valid node indices for a single element.

    Excludes fill values, returning only the actual node indices
    that define this element.

    Args:
        idx: Element index (row in the connectivity array).

    Returns:
        1D array of valid node indices for the element.
    """
    row = self.data[idx]
    if self.data.ndim == 1:
        result = np.atleast_1d(row)
    else:
        result = row[row != self.fill_value]
    return result

nodes_per_element() #

Return the number of valid nodes per element.

Returns:

Type Description
ndarray

1D integer array of length n_elements with the count
ndarray

of valid (non-fill) nodes for each element.
Source code in src/pyramids/netcdf/ugrid/connectivity.py 
115
116
117
118
119
120
121
122
123
124
125
126
def nodes_per_element(self) -> np.ndarray:
    """Return the number of valid nodes per element.

    Returns:
        1D integer array of length n_elements with the count
        of valid (non-fill) nodes for each element.
    """
    if self.data.ndim == 1:
        result = np.ones(self.n_elements, dtype=np.intp)
    else:
        result = np.sum(self.data != self.fill_value, axis=1).astype(np.intp)
    return result

is_triangular() #

Check if all elements have exactly 3 nodes.

Returns:

Type Description
bool

True if every element has exactly 3 valid nodes.
Source code in src/pyramids/netcdf/ugrid/connectivity.py 
128
129
130
131
132
133
134
135
136
def is_triangular(self) -> bool:
    """Check if all elements have exactly 3 nodes.

    Returns:
        True if every element has exactly 3 valid nodes.
    """
    counts = self.nodes_per_element()
    result = bool(np.all(counts == 3))
    return result

as_masked() #

Return a masked array with fill values masked.

Returns:

Type Description
MaskedArray

MaskedArray where positions with fill_value are masked.
Source code in src/pyramids/netcdf/ugrid/connectivity.py 
138
139
140
141
142
143
144
145
def as_masked(self) -> np.ma.MaskedArray:
    """Return a masked array with fill values masked.

    Returns:
        MaskedArray where positions with fill_value are masked.
    """
    result = np.ma.MaskedArray(self.data, mask=(self.data == self.fill_value))
    return result