Skip to content

io

InputInterface

Bases: Protocol

Common stack interface for input and output.

Such objects must export NumPy-like dtype, shape and ndim attributes.

Source code in src/spurt/io/_interface.py
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@runtime_checkable
class InputInterface(Protocol):
    """
    Common stack interface for input and output.

    Such objects must export NumPy-like `dtype`, `shape` and `ndim`
    attributes.
    """

    @property
    def dtype(self) -> np.dtype:
        """numpy.dtype : Data-type of the array's elements."""

    @property
    def shape(self) -> tuple[int, ...]:
        """tuple of int : Tuple of array dimensions."""  # noqa: D403

    @property
    def ndim(self) -> int:
        """int : Number of array dimension."""  # noqa: D403

    def __getitem__(self, key: slice | tuple[slice, ...], /) -> ArrayLike:
        """Read a block of data."""

dtype property

numpy.dtype : Data-type of the array's elements.

ndim property

int : Number of array dimension.

shape property

tuple of int : Tuple of array dimensions.

__getitem__(key)

Read a block of data.

Source code in src/spurt/io/_interface.py
39
40
def __getitem__(self, key: slice | tuple[slice, ...], /) -> ArrayLike:
    """Read a block of data."""

InputStackInterface

Bases: StackInterface, Protocol

An array-like interface for reading input datasets.

InputStackInterface defines the abstract interface that types must conform to in order to be valid inputs to `spurt.unwrap() function.

Source code in src/spurt/io/_interface.py
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@runtime_checkable
class InputStackInterface(StackInterface, Protocol):
    """
    An array-like interface for reading input datasets.

    `InputStackInterface` defines the abstract interface that types must
    conform to in order to be valid inputs to ``spurt.unwrap()` function.
    """

    def get_time_slice(self, key: int) -> ArrayLike:
        """Read a block of data in time."""

    def get_spatial_slice(self, key: int) -> ArrayLike:
        """Read a block of data in space."""

get_spatial_slice(key)

Read a block of data in space.

Source code in src/spurt/io/_interface.py
115
116
def get_spatial_slice(self, key: int) -> ArrayLike:
    """Read a block of data in space."""

get_time_slice(key)

Read a block of data in time.

Source code in src/spurt/io/_interface.py
112
113
def get_time_slice(self, key: int) -> ArrayLike:
    """Read a block of data in time."""

Irreg3DInput

Bases: InputStackInterface

A single 2D array as input.

One axis represents time and the other some flattened representation in space. time_dim is an optional input to indicate time axis.

Source code in src/spurt/io/_three_d.py
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
class Irreg3DInput(InputStackInterface):
    """A single 2D array as input.

    One axis represents time and the other some flattened representation
    in space. `time_dim` is an optional input to indicate time axis.
    """

    def __init__(
        self,
        data: InputInterface,
        xy: InputInterface,
        time_dim: int = 0,
    ):
        """Initialize a stack interface for a 2D numpy array.

        One axis represents time axis and the other spatial axis.

        Parameters
        ----------
        data: InputInterface
            2D array as input
        xy: InputInterface
            N x 2 array of indices or coordinates of each pixel in
            space, where N is the number of spatial points.
        time_dim: int, optional
            Axis of the array that represents the time dimension
        """
        # Check number of dimensions of input array
        if data.ndim != 2:
            clsname = type(self).__name__
            errmsg = (
                f"{clsname} only supports 2D arrays. Unsupported shape {data.shape}"
            )
            raise ValueError(errmsg)

        if xy.ndim != 2:
            clsname = type(self).__name__
            errmsg = f"{clsname} accepts xy as a 2D array. Unsupported shape {xy.shape}"
            raise ValueError(errmsg)

        # Check if time_dim is within limits
        if time_dim >= data.ndim:
            clsname = type(self).__name__
            errmsg = f"{clsname} received time dimension outside array dimensions"
            raise ValueError(errmsg)

        if xy.shape[0] != data.shape[1 - time_dim]:
            clsname = type(self).__name__
            errmsg = (
                f"{clsname} mismatch of data and xy arrays"
                f"{data.shape} vs {xy.shape} with time axis {time_dim}"
            )
            raise ValueError(errmsg)

        # We only hold a reference to the data
        self._data = data
        self._time_dim = time_dim
        self._xy = xy

    @property
    def data(self) -> ArrayLike:
        return self._data

    @property
    def dtype(self) -> DTypeLike:
        return self._data.dtype

    @property
    def ndim(self) -> int:
        return 2

    @property
    def time_dim(self) -> int:
        return self._time_dim

    @property
    def space_dim(self) -> int:
        return int(1 - self._time_dim)

    @property
    def _space_size(self) -> int:
        return self._data.shape[self.space_dim]

    @property
    def _time_size(self) -> int:
        return self._data.shape[self._time_dim]

    @property
    def shape(self) -> tuple[int, ...]:
        return self._data.shape

    @property
    def xcoords(self) -> ArrayLike:
        """Return the x-coordinate of the points as an array."""
        return self._xy[:, slice(0, 1)]

    @property
    def ycoords(self) -> ArrayLike:
        """Return the y-coordinate of the points as an array."""
        return self._xy[:, slice(1, 2)]

    def get_time_slice(self, key: int) -> ArrayLike:
        """Read a block of data in time.

        Only makes a copy if absolutely needed.

        Parameters
        ----------
        key: int
            Index on the time axis
        """
        ind: tuple[slice, ...] = tuple(
            slice(None) if ii != self._time_dim else slice(key, key + 1)
            for ii in range(self._data.ndim)
        )
        return np.ravel(self._data[ind])

    def get_spatial_slice(self, key: int) -> ArrayLike:
        """Read a block of data in space.

        Only makes a copy if absolutely needed.

        Parameters
        ----------
        key: int
            Index on the space axis
        """
        ind: tuple[slice, ...] = tuple(
            slice(None) if ii != self.space_dim else slice(key, key + 1)
            for ii in range(self._data.ndim)
        )
        return np.ravel(self._data[ind])

xcoords property

Return the x-coordinate of the points as an array.

ycoords property

Return the y-coordinate of the points as an array.

__init__(data, xy, time_dim=0)

Initialize a stack interface for a 2D numpy array.

One axis represents time axis and the other spatial axis.

Parameters:

Name Type Description Default
data InputInterface

2D array as input

required
xy InputInterface

N x 2 array of indices or coordinates of each pixel in space, where N is the number of spatial points.

required
time_dim int

Axis of the array that represents the time dimension

0
Source code in src/spurt/io/_three_d.py
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
def __init__(
    self,
    data: InputInterface,
    xy: InputInterface,
    time_dim: int = 0,
):
    """Initialize a stack interface for a 2D numpy array.

    One axis represents time axis and the other spatial axis.

    Parameters
    ----------
    data: InputInterface
        2D array as input
    xy: InputInterface
        N x 2 array of indices or coordinates of each pixel in
        space, where N is the number of spatial points.
    time_dim: int, optional
        Axis of the array that represents the time dimension
    """
    # Check number of dimensions of input array
    if data.ndim != 2:
        clsname = type(self).__name__
        errmsg = (
            f"{clsname} only supports 2D arrays. Unsupported shape {data.shape}"
        )
        raise ValueError(errmsg)

    if xy.ndim != 2:
        clsname = type(self).__name__
        errmsg = f"{clsname} accepts xy as a 2D array. Unsupported shape {xy.shape}"
        raise ValueError(errmsg)

    # Check if time_dim is within limits
    if time_dim >= data.ndim:
        clsname = type(self).__name__
        errmsg = f"{clsname} received time dimension outside array dimensions"
        raise ValueError(errmsg)

    if xy.shape[0] != data.shape[1 - time_dim]:
        clsname = type(self).__name__
        errmsg = (
            f"{clsname} mismatch of data and xy arrays"
            f"{data.shape} vs {xy.shape} with time axis {time_dim}"
        )
        raise ValueError(errmsg)

    # We only hold a reference to the data
    self._data = data
    self._time_dim = time_dim
    self._xy = xy

get_spatial_slice(key)

Read a block of data in space.

Only makes a copy if absolutely needed.

Parameters:

Name Type Description Default
key int

Index on the space axis

required
Source code in src/spurt/io/_three_d.py
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def get_spatial_slice(self, key: int) -> ArrayLike:
    """Read a block of data in space.

    Only makes a copy if absolutely needed.

    Parameters
    ----------
    key: int
        Index on the space axis
    """
    ind: tuple[slice, ...] = tuple(
        slice(None) if ii != self.space_dim else slice(key, key + 1)
        for ii in range(self._data.ndim)
    )
    return np.ravel(self._data[ind])

get_time_slice(key)

Read a block of data in time.

Only makes a copy if absolutely needed.

Parameters:

Name Type Description Default
key int

Index on the time axis

required
Source code in src/spurt/io/_three_d.py
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def get_time_slice(self, key: int) -> ArrayLike:
    """Read a block of data in time.

    Only makes a copy if absolutely needed.

    Parameters
    ----------
    key: int
        Index on the time axis
    """
    ind: tuple[slice, ...] = tuple(
        slice(None) if ii != self._time_dim else slice(key, key + 1)
        for ii in range(self._data.ndim)
    )
    return np.ravel(self._data[ind])

OutputInterface

Bases: Protocol

An array-like interface for writing output datasets.

OutputInterface must export NumPy-like dtype, shape, and ndim attributes and must support NumPy-style slice-based indexing.

Source code in src/spurt/io/_interface.py
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@runtime_checkable
class OutputInterface(Protocol):
    """
    An array-like interface for writing output datasets.

    `OutputInterface` must export NumPy-like `dtype`, `shape`, and `ndim`
    attributes and must support NumPy-style slice-based indexing.
    """

    @property
    def dtype(self) -> np.dtype:
        """numpy.dtype : Data-type of the array's elements."""

    @property
    def shape(self) -> tuple[int, ...]:
        """tuple of int : Tuple of array dimensions."""  # noqa: D403

    @property
    def ndim(self) -> int:
        """int : Number of array dimensions."""  # noqa: D403

    def __setitem__(self, key: slice | tuple[slice, ...], value: np.ndarray, /) -> None:
        """Write a block of data."""

dtype property

numpy.dtype : Data-type of the array's elements.

ndim property

int : Number of array dimensions.

shape property

tuple of int : Tuple of array dimensions.

__setitem__(key, value)

Write a block of data.

Source code in src/spurt/io/_interface.py
64
65
def __setitem__(self, key: slice | tuple[slice, ...], value: np.ndarray, /) -> None:
    """Write a block of data."""

OutputStackInterface

Bases: StackInterface, Protocol

An array-like interface for writing output datasets.

OutputStackInterface defines the abstract interface that types must conform to in order to be valid outputs of the `spurt.unwrap() function.

Source code in src/spurt/io/_interface.py
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@runtime_checkable
class OutputStackInterface(StackInterface, Protocol):
    """
    An array-like interface for writing output datasets.

    `OutputStackInterface` defines the abstract interface that types must
    conform to in order to be valid outputs of the ``spurt.unwrap()` function.
    """

    def set_time_slice(self, key: int, array: ArrayLike) -> None:
        """Write a block of data in time."""

    def set_spatial_slice(self, key: int, array: ArrayLike) -> None:
        """Write a block of data in space."""

set_spatial_slice(key, array)

Write a block of data in space.

Source code in src/spurt/io/_interface.py
131
132
def set_spatial_slice(self, key: int, array: ArrayLike) -> None:
    """Write a block of data in space."""

set_time_slice(key, array)

Write a block of data in time.

Source code in src/spurt/io/_interface.py
128
129
def set_time_slice(self, key: int, array: ArrayLike) -> None:
    """Write a block of data in time."""

Reg3DInput

Bases: InputStackInterface

A single 3D array as input.

A 2D image will just be a 3D array with size of time axis set to 1. This is largely a helper class to assist in development. time_dim is an optional input to indicate time axis.

Source code in src/spurt/io/_three_d.py
 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
 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
class Reg3DInput(InputStackInterface):
    """A single 3D array as input.

    A 2D image will just be a 3D array with size of time axis set to 1. This is
    largely a helper class to assist in development. `time_dim` is an optional
    input to indicate time axis.
    """

    def __init__(
        self,
        data: InputInterface,
        time_dim: int = 0,
    ):
        """Initialize a stack interface for a numpy array.

        Representing 3D data in a cube - regularly spaced in spatial dimension.

        Parameters
        ----------
        data: InputInterface
            3D array as input
        time_dim: int, optional
            Axis of the array that represents the time dimension
        """
        # Check number of dimensions of input array
        if data.ndim != 3:
            clsname = type(self).__name__
            errmsg = (
                f"{clsname} only supports 3D arrays. Unsupported shape {data.shape}"
            )
            raise ValueError(errmsg)

        # Check if time_dim is within limits
        if time_dim >= data.ndim:
            clsname = type(self).__name__
            errmsg = f"{clsname} received time dimension outside array dimensions"
            raise ValueError(errmsg)

        # We only hold a reference to the data
        self._data = data
        self._time_dim = time_dim

    @property
    def data(self) -> ArrayLike:
        return self._data

    @property
    def dtype(self) -> DTypeLike:
        return self._data.dtype

    @property
    def ndim(self) -> int:
        return 2

    @property
    def time_dim(self) -> int:
        """We flatten spatial dimensions."""
        return int(self._time_dim > 0)

    @property
    def space_dim(self) -> int:
        """We flatten spatial dimensions."""
        return int(self._time_dim == 0)

    @property
    def _space_shape(self) -> tuple[int, ...]:
        return tuple(
            ss for ii, ss in enumerate(self._data.shape) if ii != self._time_dim
        )

    @property
    def _space_size(self) -> int:
        return int(np.prod(self._space_shape))

    @property
    def _time_size(self) -> int:
        return self._data.shape[self._time_dim]

    @property
    def shape(self) -> tuple[int, ...]:
        if self.time_dim:
            return (self._space_size, self._time_size)

        return (self._time_size, self._space_size)

    @property
    def xcoords(self) -> ArrayLike:
        """Return the x-coordinate of the points as an array.

        This is to mimic an irregular grid.
        """
        return np.arange(np.prod(self._data.shape), dtype=int) % self._space_shape[1]

    @property
    def ycoords(self) -> ArrayLike:
        """Return the y-coordinate of the points as an array.

        This is to mimic an irregular grid.
        """
        return np.arange(np.prod(self._data.shape), dtype=int) // self._space_shape[1]

    def get_time_slice(self, key: int) -> ArrayLike:
        """Read a block of data in time.

        Only makes a copy if absolutely needed.
        """
        ind: slice | tuple[slice, ...] = tuple(
            slice(None) if ii != self._time_dim else slice(key, key + 1)
            for ii in range(self._data.ndim)
        )
        return np.ravel(self._data[ind])

    def get_spatial_slice(self, key: int) -> ArrayLike:
        """Read a block of data in space.

        Only makes a copy if absolutely needed.
        """
        ind: list[slice] = [
            slice(x, x + 1) for x in np.unravel_index(key, self._space_shape)
        ]
        ind.insert(self._time_dim, slice(None))

        return np.ravel(self._data[tuple(ind)])

space_dim property

We flatten spatial dimensions.

time_dim property

We flatten spatial dimensions.

xcoords property

Return the x-coordinate of the points as an array.

This is to mimic an irregular grid.

ycoords property

Return the y-coordinate of the points as an array.

This is to mimic an irregular grid.

__init__(data, time_dim=0)

Initialize a stack interface for a numpy array.

Representing 3D data in a cube - regularly spaced in spatial dimension.

Parameters:

Name Type Description Default
data InputInterface

3D array as input

required
time_dim int

Axis of the array that represents the time dimension

0
Source code in src/spurt/io/_three_d.py
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
def __init__(
    self,
    data: InputInterface,
    time_dim: int = 0,
):
    """Initialize a stack interface for a numpy array.

    Representing 3D data in a cube - regularly spaced in spatial dimension.

    Parameters
    ----------
    data: InputInterface
        3D array as input
    time_dim: int, optional
        Axis of the array that represents the time dimension
    """
    # Check number of dimensions of input array
    if data.ndim != 3:
        clsname = type(self).__name__
        errmsg = (
            f"{clsname} only supports 3D arrays. Unsupported shape {data.shape}"
        )
        raise ValueError(errmsg)

    # Check if time_dim is within limits
    if time_dim >= data.ndim:
        clsname = type(self).__name__
        errmsg = f"{clsname} received time dimension outside array dimensions"
        raise ValueError(errmsg)

    # We only hold a reference to the data
    self._data = data
    self._time_dim = time_dim

get_spatial_slice(key)

Read a block of data in space.

Only makes a copy if absolutely needed.

Source code in src/spurt/io/_three_d.py
128
129
130
131
132
133
134
135
136
137
138
def get_spatial_slice(self, key: int) -> ArrayLike:
    """Read a block of data in space.

    Only makes a copy if absolutely needed.
    """
    ind: list[slice] = [
        slice(x, x + 1) for x in np.unravel_index(key, self._space_shape)
    ]
    ind.insert(self._time_dim, slice(None))

    return np.ravel(self._data[tuple(ind)])

get_time_slice(key)

Read a block of data in time.

Only makes a copy if absolutely needed.

Source code in src/spurt/io/_three_d.py
117
118
119
120
121
122
123
124
125
126
def get_time_slice(self, key: int) -> ArrayLike:
    """Read a block of data in time.

    Only makes a copy if absolutely needed.
    """
    ind: slice | tuple[slice, ...] = tuple(
        slice(None) if ii != self._time_dim else slice(key, key + 1)
        for ii in range(self._data.ndim)
    )
    return np.ravel(self._data[ind])