pikers
/
piker
3
0
Fork 0
Code Issues 4 Pull Requests 6 Packages Projects Releases Wiki Activity

Facepalm**2: fix array-read-slice, like actually.. 

We need to subtract the first index in the array segment read, not the
first index value in the time-sliced output, to get the correct offset
into the non-absolute (`ShmArray.array` read) array..

Further we **do** need the `&` between the advance indexing conditions
and this adds profiling to see that it is indeed real slow (like 20ms
ish even when using `np.where()`).
epoch_index_backup
Tyler Goodlet 2022-12-02 18:49:12 -05:00
parent a2f75a83b6
commit 2e6f14afb3
1 changed files with 25 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
piker/data/flows.py
View File

@ -48,6 +48,10 @@ from ._sharedmem import (
from ._sampling import ( from ._sampling import (
iter_ohlc_periods, iter_ohlc_periods,
) )
from .._profile import (
Profiler,
pg_profile_enabled,
)
if TYPE_CHECKING: if TYPE_CHECKING:
from pyqtgraph import PlotItem from pyqtgraph import PlotItem
@ -264,6 +268,13 @@ class Flume(Struct):
for the caller to use to slice the input array if needed. for the caller to use to slice the input array if needed.
''' '''
profiler = Profiler(
msg='Flume.slice_from_time()',
disabled=not pg_profile_enabled(),
ms_threshold=4,
# ms_threshold=ms_slower_then,
)
times = arr['time'] times = arr['time']
index = arr['index'] index = arr['index']
@ -285,12 +296,12 @@ class Flume(Struct):
# use advanced indexing to map the # use advanced indexing to map the
# time range to the index range. # time range to the index range.
mask: np.ndarray = ( mask: np.ndarray = np.where(
(times >= start_t) (times >= start_t)
| # fml, i guess it's not an `&` ?? &
(times < stop_t) (times < stop_t)
) )
profiler('advanced indexing slice')
# TODO: if we can ensure each time field has a uniform # TODO: if we can ensure each time field has a uniform
# step we can instead do some arithmetic to determine # step we can instead do some arithmetic to determine
# the equivalent index like we used to? # the equivalent index like we used to?
@ -302,6 +313,8 @@ class Flume(Struct):
i_by_t = index[mask] i_by_t = index[mask]
try: try:
i_0 = i_by_t[0] i_0 = i_by_t[0]
i_last = i_by_t[-1]
i_first_read = index[0]
except IndexError: except IndexError:
if ( if (
start_t < times[0] start_t < times[0]
@ -319,17 +332,20 @@ class Flume(Struct):
None, None,
) )
abs_slc = slice( abs_slc = slice(i_0, i_last)
i_0,
i_by_t[-1],
)
# slice data by offset from the first index # slice data by offset from the first index
# available in the passed datum set. # available in the passed datum set.
read_slc = slice( read_slc = slice(
i_0 - index[0], i_0 - i_first_read,
i_by_t[-1] - i_0, i_last - i_first_read + 1,
) )
profiler(
'slicing complete'
f'{start_t} -> {abs_slc.start} | {read_slc.start}\n'
f'{stop_t} -> {abs_slc.stop} | {read_slc.stop}\n'
)
# also return the readable data from the timerange # also return the readable data from the timerange
return ( return (
abs_slc, abs_slc,