Repository navigation
Expand file tree
/
Copy path_base.py
More file actions
809 lines (645 loc) · 29.2 KB
/
Copy path_base.py
File metadata and controls
809 lines (645 loc) · 29.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
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
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
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
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
from __future__ import annotations
from collections.abc import Callable, Sequence
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
import inspect
from numbers import Real
from pprint import pformat
import textwrap
from typing import Any, TYPE_CHECKING
import numpy as np
from numpy.typing import ArrayLike
from ...utils import ArrayProtocol, FutureProtocol, CudaArrayProtocol
from ...graphics import Graphic
from ._async import run_in_thread_pool, run_sync, wait_for_future
if TYPE_CHECKING:
from ._ndw_subplot import NDWSubplot
# must take arguments: array-like, `axis`: int, `keepdims`: bool
WindowFuncCallable = Callable[[ArrayLike, int, bool], ArrayLike]
def identity(index: int) -> int:
return round(index)
class NDProcessor:
def __init__(
self,
data: ArrayProtocol,
dims: Sequence[str],
spatial_dims: Sequence[str] | None,
slider_dim_transforms: dict[str, Callable[[Any], int] | ArrayLike] = None,
window_funcs: dict[
str, tuple[WindowFuncCallable | None, int | float | None]
] = None,
window_order: tuple[str, ...] = None,
spatial_func: Callable[[ArrayProtocol], ArrayProtocol] | None = None,
):
"""
Base class for managing n-dimensional data and producing array slices.
Wraps array-like ``data`` and provides an interface for indexing slider dimensions, applying window functions,
spatial functions, and mapping reference-space values to local array indices. Subclasses must implement
:meth:`get`, which is called when the :class:`ReferenceIndex` updates.
Subclasses can implement any type of data representation, they do not necessarily need to be array-like.
However their ``get()`` method must still return a data slice that corresponds to the graphical representation
they map to.
Every dimension that is *not* listed in ``spatial_dims`` becomes a slider
dimension. Each slider dim must have a ``ReferenceRange`` defined in the
``ReferenceIndex`` of the parent ``NDWidget``. The widget uses this to direct
a change in the ``ReferenceIndex`` and update the graphics.
Parameters
----------
data: ArrayProtocol
data object that is managed, usually uses the ArrayProtocol. Custom subclasses can manage any kind of data
object but the corresponding :meth:`get` must return an array-like that maps to a graphical representation.
dims: Sequence[str]
names for each dimension in ``data``. Dimensions not listed in
``spatial_dims`` are treated as slider dimensions and **must** appear as
keys in the parent ``NDWidget``'s ``ref_ranges``
Examples::
``("time", "depth", "row", "col")``
``("channels", "time", "xy")``
``("keypoints", "time", "xyz")``
A custom subclass's ``data`` object doesn't necessarily need to have these dims, but the ``get()`` method
must operate as if these dimensions exist and return an array that matches the spatial dimensions.
spatial_dims: Sequence[str]
Subset of ``dims`` that are spatial (rendered) dimensions **in display order**. All remaining dims are
treated as slider dims. See subclass for specific info.
slider_dim_transforms: dict mapping dim_name -> Callable, an ArrayLike, or None
Per-slider-dim mapping from reference-space values to local array indices.
You may also provide an array of reference values for the slider dims, ``searchsorted`` is then used
as the transform (ex: a timestamps array).
If ``None`` and identity mapping is used, i.e. rounds the current reference index value to the nearest
integer for array indexing.
If a transform is not provided for a dim then the identity mapping is used.
window_funcs: dict[
str, tuple[WindowFuncCallable | None, int | float | None]
]
Per-slider-dim window functions applied around the current slider position. Ex: {"time": (np.mean, 2.5)}.
Each value is a ``(func, window_size)`` pair where:
* *func* must accept ``axis: int`` and ``keepdims: bool`` kwargs
(ex: ``np.mean``, ``np.max``). The window function **must** return an array that has the same dimensions
as specified in the NDProcessor, therefore the size of any dim along which a window_func was applied
should reduce to ``1``. These dims must not be removed by the window_func.
* *window_size* is in reference-space units (ex: 2.5 seconds).
window_order: tuple[str, ...]
Order in which window functions are applied across dims. Only dims listed
here have their window function applied. window_funcs are ignored for any
dims not specified in ``window_order``
spatial_func:
A function applied to the spatial slice *after* window_funcs right before rendering.
"""
dims = tuple(dims)
if not all([isinstance(d, str) for d in dims]):
raise TypeError
self._dims = dims
self.data = data
self.spatial_dims = spatial_dims
self.slider_dim_transforms = slider_dim_transforms
self.window_funcs = window_funcs
self.window_order = window_order
self.spatial_func = spatial_func
# window_funcs and spatial_func are dispatched with an executor so they don't block the rendercanvas loop.
# CUDA arrays run directly since they are inherently async already, the user is expected to provide CUDA
# functions if the data arrays are CUDA (ex: torch functions, not numpy functions)
self._executor = ThreadPoolExecutor(
max_workers=1, thread_name_prefix=f"ndp-{id(self):x}"
)
def close(self):
"""Shut down the thread pool."""
self._executor.shutdown(wait=False, cancel_futures=True)
@property
def data(self) -> ArrayProtocol:
"""
get or set managed data. If setting with new data, the new data is interpreted
to have the same dims (i.e. same dim names and ordering of dims).
"""
return self._data
@data.setter
def data(self, data: ArrayProtocol):
# data can be set, but the dims must still match/have the same meaning
if data is None:
# we allow data to be None, in this case no ndgraphic is rendered
# useful when we want to initialize an NDWidget with no traces for example
# and populate it as components/channels are selected
self._data = None
return
if not isinstance(data, ArrayProtocol):
# check for general array-like requirements
raise TypeError("`data` must implement the ArrayProtocol")
if data.ndim != len(self.dims):
raise IndexError("must specify a dim for every dimension in the data array")
self._data = data
@property
def shape(self) -> dict[str, int]:
"""interpreted shape of the data"""
return {d: n for d, n in zip(self.dims, self.data.shape)}
@property
def ndim(self) -> int:
"""number of dims"""
return self.data.ndim
@property
def dims(self) -> tuple[str, ...]:
"""dim names, **ordered as laid out in the array**"""
# these are read-only and cannot be set after it's created
# the user should create a new NDGraphic if they need different dims
# I can't think of a use case where we'd want to change the dims, and
# I think that would be complicated and probably and anti-pattern
return self._dims
@property
def spatial_dims(self) -> tuple[str, ...]:
"""Spatial dims, **in display order**"""
return self._spatial_dims
@spatial_dims.setter
def spatial_dims(self, sdims: Sequence[str]):
for dim in sdims:
if dim not in self.dims:
raise KeyError
self._spatial_dims = tuple(sdims)
@property
def tooltip(self) -> bool:
"""
whether or not a custom tooltip formatter method exists
"""
return False
def tooltip_format(self, *args) -> str | None:
"""
Override in subclass to format custom tooltips
"""
return None
@property
def slider_dims(self) -> set[str]:
"""Slider dim names, ``set(dims) - set(spatial_dims), **unordered**"""
return set(self.dims) - set(self.spatial_dims)
@property
def n_slider_dims(self):
"""number of slider dims, i.e. len(slider_dims)"""
return len(self.slider_dims)
@property
def window_funcs(
self,
) -> dict[str, tuple[WindowFuncCallable | None, int | float | None]]:
"""get or set window functions, see docstring for details"""
return self._window_funcs
@window_funcs.setter
def window_funcs(
self,
window_funcs: (
dict[str, tuple[WindowFuncCallable | None, int | float | None] | None]
| None
),
):
if window_funcs is None:
# tuple of (None, None) makes the checks easier in _apply_window_funcs
self._window_funcs = {d: (None, None) for d in self.slider_dims}
return
for k in window_funcs.keys():
if k not in self.slider_dims:
raise KeyError
func = window_funcs[k][0]
size = window_funcs[k][1]
if func is None:
pass
elif callable(func):
sig = inspect.signature(func)
if "axis" not in sig.parameters or "keepdims" not in sig.parameters:
raise TypeError(
f"Each window function must take an `axis` and `keepdims` argument, "
f"you passed: {func} with the following function signature: {sig}"
)
else:
raise TypeError(
f"`window_funcs` must be a dict mapping dim names to a tuple of the window function callable and "
f"window size, {'name': (func, size), ...}.\nYou have passed: {window_funcs}"
)
if size is None:
pass
elif not isinstance(size, Real):
raise TypeError
elif size < 0:
raise ValueError
# fill in rest with None
for d in self.slider_dims:
if d not in window_funcs.keys():
window_funcs[d] = (None, None)
self._window_funcs = window_funcs
@property
def window_order(self) -> tuple[str, ...]:
"""get or set dimension order in which window functions are applied"""
return self._window_order
@window_order.setter
def window_order(self, order: tuple[str] | None):
if order is None:
self._window_order = tuple()
return
if not set(order).issubset(self.slider_dims):
raise ValueError(
f"each dimension in `window_order` must be a slider dim. You passed order: {order} "
f"and the slider dims are: {self.slider_dims}"
)
self._window_order = tuple(order)
@property
def spatial_func(self) -> Callable[[ArrayProtocol], ArrayProtocol] | None:
"""get or set the spatial function which is applied on the data slice after the window functions"""
return self._spatial_func
@spatial_func.setter
def spatial_func(
self, func: Callable[[ArrayProtocol], ArrayProtocol]
) -> Callable | None:
if not callable(func) and func is not None:
raise TypeError
self._spatial_func = func
@property
def slider_dim_transforms(self) -> dict[str, Callable[[Any], int]]:
"""get or set the slider_dim_transforms, see docstring for details"""
return self._index_mappings
@slider_dim_transforms.setter
def slider_dim_transforms(
self, maps: dict[str, Callable[[Any], int] | ArrayLike | None] | None
):
if maps is None:
self._index_mappings = {d: identity for d in self.dims}
return
for d in maps.keys():
if d not in self.dims:
raise KeyError(
f"`index_mapping` provided for non-existent dimension: {d}, existing dims are: {self.dims}"
)
if isinstance(maps[d], ArrayProtocol):
# create a searchsorted mapping function automatically
maps[d] = maps[d].searchsorted
elif maps[d] is None:
# assign identity mapping
maps[d] = identity
for d in self.dims:
# fill in any unspecified maps with identity
if d not in maps.keys():
maps[d] = identity
self._index_mappings = maps
def _ref_index_to_array_index(self, dim: str, ref_index: Any) -> int:
# wraps slider_dim_transforms, clamps between 0 and the array size in this dim
# ref-space -> local-array-index transform
index = self.slider_dim_transforms[dim](ref_index)
# clamp between 0 and array size in this dim
return max(min(index, self.shape[dim] - 1), 0)
def _get_slider_dims_indexer(self, indices: dict[str, Any]) -> dict[str, slice]:
"""
Creates an indexer dict mapping each slider_dim -> slice object.
- If a window_func is defined for a dim and the dim appears in ``window_order``,
the slice is defined as:
start: index - half_window
stop: index + half_window
step: 1
It then applies the slider_dim_transform to the start and stop to map these values from reference-space to
the local array index, and then finally produces the slice object in local array indices.
ex: if we have indices = {"time": 50.0}, a window size of 5.0s and the ``slider_dim_transform``
for time is based on a sampling rate of 10Hz, the window in ref units is [45.0, 55.0], and the final
slice object would be ``slice(450, 550, 1)``.
- If no window func is specified, the final slice just corresponds to that index as an int array-index.
This exists separate from ``_apply_window_functions()`` because it is useful for debugging purposes.
Parameters
----------
indices : dict[str, Any], {dim: ref_value}
Reference-space values for each slider dim. Must contain an entry
for every slider dim; raises ``IndexError`` otherwise.
ex: {"time": 46.397, "depth": 23.24}
Returns
-------
dict[str, slice]
Indexer compatible for ``xr.DataArray.isel()``, with one ``slice`` per
slider dim. These are array indices mapped from the reference space using
the given ``slider_dim_transform``.
Raises
------
IndexError
If ``indices`` are not provided for every ``slider_dim``
"""
if set(indices.keys()) != set(self.slider_dims):
raise IndexError(
f"Must provide an index for all slider dims: {self.slider_dims}, you have provided: {indices.keys()}"
)
indexer = dict()
# get only slider dims which are not also spatial dims (example: p dim for positional data)
# since `p` dim windowing is dealt with separately for positional data
slider_dims = set(self.slider_dims) - set(self.spatial_dims)
# go through each slider dim and accumulate slice objects
for dim in slider_dims:
# index for this dim in reference space
index_ref = indices[dim]
if dim not in self.window_funcs.keys():
wf, ws = None, None
else:
# get window func and size in reference units
wf, ws = self.window_funcs[dim]
# if a window function exists for this dim, and it's specified in the window order
if (wf is not None) and (ws is not None) and (dim in self.window_order):
# half window in reference units
hw = ws / 2
# start in reference units
start_ref = index_ref - hw
# stop in ref units
stop_ref = index_ref + hw
# map start and stop ref to array indices
start = self.slider_dim_transforms[dim](start_ref)
stop = self.slider_dim_transforms[dim](stop_ref)
# clamp within array bounds
start = max(min(self.shape[dim] - 1, start), 0)
stop = max(min(self.shape[dim] - 1, stop), 0)
indexer[dim] = slice(start, stop, 1)
else:
# no window func for this dim, direct indexing
# index mapped to array index
index = self.slider_dim_transforms[dim](index_ref)
# clamp within the bounds
start = max(min(self.shape[dim] - 1, index), 0)
# stop index is just the start index + 1
indexer[dim] = slice(start, start + 1, 1)
return indexer
async def _apply_window_functions(
self, windowed_array: ArrayProtocol
) -> ArrayProtocol:
"""
apply window functions in the order specified by
``window_order``.
For numpy arrays each func is dispatched to the per-processor thread pool so it
does not block the rendercanvas event loop. CUDA arrays are run directly since
cuda functions (ex: torch) are already async.
Parameters
----------
windowed_array: ArrayProtocol
array that has been sliced with the desired windows at an index
Returns
-------
ArrayProtocol
Data slice after windowed indexing and window function application,
with the same dims as the original data. Dims of size ``1`` are not
squeezed.
"""
# apply window funcs in the specified order
for dim in self.window_order:
if self.window_funcs[dim] is None:
continue
func, _ = self.window_funcs[dim]
axis = self.dims.index(dim)
# ``keepdims=True`` is critical, any "collapsed" dims will be of size ``1``.
# Ex: if `array` is of shape [10, 512, 512] and we applied the np.mean() window func on the first dim
# ``keepdims`` means the resultant shape is [1, 512, 512] and NOT [512, 512]
# this is necessary for applying window functions on multiple dims separately and so that the
# dims names correspond after all the window funcs are applied.
if isinstance(windowed_array, CudaArrayProtocol):
windowed_array = func(windowed_array, axis=axis, keepdims=True)
else:
windowed_array = await run_in_thread_pool(
self._executor, func, windowed_array, axis=axis, keepdims=True
)
return windowed_array
async def get_window_output(self, indices: dict[str, Any]) -> ArrayProtocol:
"""
Applies any window functions and returns squeezed sliced array transposed in the order of the given spatial dims
Parameters
----------
indices
Returns
-------
"""
# windowed slice if user set any window funcs
windowed_slice = await self._get_raw_data_slice(indices)
# convert to numpy array; CUDA arrays pass through and are converted at the end of the pipeline
if not isinstance(windowed_slice, CudaArrayProtocol):
windowed_slice = np.asarray(windowed_slice)
# apply window funcs
if len(self.slider_dims) > 0:
windowed_slice = await self._apply_window_functions(windowed_slice)
# squeeze out all slider dims which should now be size 1
# set(dims) - set(spatial_dims) since some spatial dims can also be slider, so get only pure non-spatial dims
slider_dims_int = tuple(
self.dims.index(d) for d in set(self.dims) - set(self.spatial_dims)
)
windowed_slice = windowed_slice.squeeze(axis=slider_dims_int)
if windowed_slice.ndim != len(self.spatial_dims):
raise ValueError(
f"windowed_slice.ndim != len(self.spatial_dims): {windowed_slice.ndim} != {len(self.spatial_dims)}"
)
# transpose to spatial dims
spatial_dims_int = tuple(
self.spatial_dims.index(d) for d in self.dims if d in self.spatial_dims
)
return windowed_slice.transpose(*spatial_dims_int)
async def _get_raw_data_slice(self, indices: dict[str, Any]) -> ArrayProtocol:
"""
Base implementation to get the raw data slice from the wrapped array.
Awaits any ``FutureProtocol`` returned by the underlying loader. CUDA arrays
are returned as-is and converted to numpy at the end of the pipeline.
"""
if len(self.slider_dims) > 0:
indexer = self._get_slider_dims_indexer(indices)
# get the data slice w.r.t. the desired windows
index_tuple = tuple(indexer.get(dim, slice(None)) for dim in self.dims)
raw_slice = self.data[index_tuple]
else:
# return everything directly
# request a slice of everything with [:] so that any data fetching, compute, etc. is actually done
raw_slice = self.data[:]
if isinstance(raw_slice, FutureProtocol):
return await wait_for_future(raw_slice)
return raw_slice
async def get(self, indices: dict[str, Any]) -> ArrayProtocol:
raise NotImplementedError
# TODO: html and pretty text repr #
# def _repr_html_(self) -> str:
# return ndp_fmt_html(self)
#
# def _repr_mimebundle_(self, **kwargs) -> dict:
# return {
# "text/plain": self._repr_text_(),
# "text/html": self._repr_html_(),
# }
def _repr_text_(self):
if self.data is None:
return f"{self.__class__.__name__}\n" f"data is None, dims: {self.dims}"
tab = "\t"
wf = {k: v for k, v in self.window_funcs.items() if v != (None, None)}
r = (
f"{self.__class__.__name__}\n"
f"shape:\n\t{self.shape}\n"
f"dims:\n\t{self.dims}\n"
f"spatial_dims:\n\t{self.spatial_dims}\n"
f"slider_dims:\n\t{self.slider_dims}\n"
f"slider_dim_transforms:\n{textwrap.indent(pformat(self.slider_dim_transforms, width=120), prefix=tab)}\n"
)
if len(wf) > 0:
r += (
f"window_funcs:\n{textwrap.indent(pformat(wf, width=120), prefix=tab)}\n"
f"window_order:\n\t{self.window_order}\n"
)
if self.spatial_func is not None:
r += f"spatial_func:\n\t{self.spatial_func}\n"
return r
class NDGraphic:
def __init__(
self,
nd_subplot: NDWSubplot,
name: str | None,
):
self._nd_subplot = nd_subplot
self._name = name
self._graphic: Graphic | None = None
# used to indicate that the NDGraphic should ignore any requests to update the indices.
# used by block_indices_ctx context manager, usecase is when the LinearSelector on timeseries
# NDGraphic changes the selection, it shouldn't change the graphic that it is on top of! Would
# also cause recursion. ReferenceIndex._render_indices checks this flag at scheduling time.
self._block_indices = False
# user settable bool to make the graphic unresponsive to change in the ReferenceIndex
self._pause = False
# the indices that current graphic data reflects
self._last_indices = None
async def _create_graphic(self):
raise NotImplementedError
@property
def pause(self) -> bool:
"""if True, changes in the reference until it is set back to False"""
return self._pause
@pause.setter
def pause(self, val: bool):
self._pause = bool(val)
@property
def name(self) -> str | None:
"""name given to the NDGraphic"""
return self._name
@property
def processor(self) -> NDProcessor:
raise NotImplementedError
@property
def graphic(self) -> Graphic:
raise NotImplementedError
@property
def indices_displayed(self) -> dict[str, Any]:
"""the indices that the graphic currently represents"""
return self._last_indices
@property
def indices(self) -> dict[str, Any]:
raise NotImplementedError
async def _set_indices_(self, indices: dict[str, Any] = None):
"""
Get the data slice for the index from the processor and write it to the graphic.
If indices is None, it uses the latest indices from the ReferenceIndex. Otherwise it uses the
indices passed when the update was scheduled.
Semi-private: only ``ReferenceIndex`` should call this. _create_graphic uses `run_sync`
to run it sync
"""
pass
# aliases for easier access to processor properties
@property
def data(self) -> Any:
"""
get or set managed data. If setting with new data, the new data is interpreted
to have the same dims (i.e. same dim names and ordering of dims).
"""
return self.processor.data
@data.setter
def data(self, data: Any):
self.processor.data = data
# create a new graphic when data has changed
if self.graphic is not None:
# it is already None if NDGraphic was initialized with no data
self._nd_subplot.subplot.delete_graphic(self.graphic)
self._graphic = None
run_sync(self._create_graphic())
# force a render
run_sync(self._set_indices_())
@property
def shape(self) -> dict[str, int]:
"""interpreted shape of the data"""
return self.processor.shape
@property
def ndim(self) -> int:
"""number of dims"""
return self.processor.ndim
@property
def dims(self) -> tuple[str, ...]:
"""dim names"""
return self.processor.dims
@property
def spatial_dims(self) -> tuple[str, ...]:
# number of spatial dims for positional data is always 3
# for image is 2 or 3, so it must be implemented in subclass
raise NotImplementedError
@property
def slider_dims(self) -> set[str]:
"""the slider dims"""
return self.processor.slider_dims
@property
def slider_dim_transforms(self) -> dict[str, Callable[[Any], int]]:
return self.processor.slider_dim_transforms
@slider_dim_transforms.setter
def slider_dim_transforms(
self, maps: dict[str, Callable[[Any], int] | ArrayLike | None] | None
):
"""get or set the slider_dim_transforms, see docstring for details"""
self.processor.slider_dim_transforms = maps
# force a render
run_sync(self._set_indices_())
@property
def window_funcs(
self,
) -> dict[str, tuple[WindowFuncCallable | None, int | float | None]]:
"""get or set window functions, see docstring for details"""
return self.processor.window_funcs
@window_funcs.setter
def window_funcs(
self,
window_funcs: (
dict[str, tuple[WindowFuncCallable | None, int | float | None] | None]
| None
),
):
self.processor.window_funcs = window_funcs
# force a render
run_sync(self._set_indices_())
@property
def window_order(self) -> tuple[str, ...]:
"""get or set dimension order in which window functions are applied"""
return self.processor.window_order
@window_order.setter
def window_order(self, order: tuple[str] | None):
self.processor.window_order = order
# force a render
run_sync(self._set_indices_())
@property
def spatial_func(self) -> Callable[[ArrayProtocol], ArrayProtocol] | None:
"""get or set the spatial_func, see docstring for details"""
return self.processor.spatial_func
@spatial_func.setter
def spatial_func(
self, func: Callable[[ArrayProtocol], ArrayProtocol]
) -> Callable | None:
"""get or set the spatial_func, see docstring for details"""
self.processor.spatial_func = func
# force a render
run_sync(self._set_indices_())
# def _repr_text_(self) -> str:
# return ndg_fmt_text(self)
#
# def _repr_html_(self) -> str:
# return ndg_fmt_html(self)
#
# def _repr_mimebundle_(self, **kwargs) -> dict:
# return {
# "text/plain": self._repr_text_(),
# "text/html": self._repr_html_(),
# }
def _repr_text_(self):
return (
f"graphic: {self.graphic.__class__.__name__}\n"
f"processor:\n{self.processor}"
)
@contextmanager
def block_indices_ctx(*ndgraphics: NDGraphic):
"""
Context manager for pausing NDGraphics from updating indices
"""
for ndg in ndgraphics:
ndg._block_indices = True
try:
yield
except Exception as e:
raise e from None # indices setter has raised, the line above and the lines below are probably more relevant!
finally:
for ndg in ndgraphics:
ndg._block_indices = False