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

Incrementally update flattend OHLC data 

After much effort (and exhaustion) but failure to get a view into our
`numpy` OHLC struct-array, this instead allocates an in-thread-memory
array which is updated with flattened data every flow update cycle.

I need to report what I think is a bug to `numpy` core about the whole
view thing not working but, more or less this gets the same behaviour
and minimizes work to flatten the sampled data for line-graphics drawing
thus improving refresh latency when drawing large downsampled curves.

Update the OHLC ds curve with view aware data sliced out from the
pre-allocated and incrementally updated data (we had to add a last index
var `._iflat` to track appends - this should be moved into a renderer
eventually?).
pre_flow
Tyler Goodlet 2022-05-10 17:57:14 -04:00
parent a05566ab53
commit 9a716de36d
1 changed files with 64 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

99
piker/ui/_flows.py
View File

@ -24,7 +24,6 @@ incremental update.
''' '''
from __future__ import annotations from __future__ import annotations
from functools import partial from functools import partial
import time
from typing import ( from typing import (
Optional, Optional,
Callable, Callable,
@ -38,7 +37,7 @@ from PyQt5.QtGui import QPainterPath
from ..data._sharedmem import ( from ..data._sharedmem import (
ShmArray, ShmArray,
# attach_shm_array open_shm_array,
) )
from .._profile import pg_profile_enabled, ms_slower_then from .._profile import pg_profile_enabled, ms_slower_then
from ._ohlc import ( from ._ohlc import (
@ -152,6 +151,7 @@ class Flow(msgspec.Struct): # , frozen=True):
render: bool = True # toggle for display loop render: bool = True # toggle for display loop
flat: Optional[ShmArray] = None flat: Optional[ShmArray] = None
x_basis: Optional[np.ndarray] = None x_basis: Optional[np.ndarray] = None
_iflat: int = 0
_last_uppx: float = 0 _last_uppx: float = 0
_in_ds: bool = False _in_ds: bool = False
@ -344,6 +344,7 @@ class Flow(msgspec.Struct): # , frozen=True):
graphics = self.graphics graphics = self.graphics
if isinstance(graphics, BarItems): if isinstance(graphics, BarItems):
fields = ['open', 'high', 'low', 'close']
# if no source data renderer exists create one. # if no source data renderer exists create one.
r = self._src_r r = self._src_r
if not r: if not r:
@ -357,17 +358,37 @@ class Flow(msgspec.Struct): # , frozen=True):
# create a flattened view onto the OHLC array # create a flattened view onto the OHLC array
# which can be read as a line-style format # which can be read as a line-style format
# shm = self.shm shm = self.shm
# self.flat = shm.unstruct_view(['open', 'high', 'low', 'close'])
# flat = self.flat = self.shm.unstruct_view(fields)
self.flat = self.shm.ustruct(fields)
self._iflat = self.shm._last.value
# import pdbpp # import pdbpp
# pdbpp.set_trace() # pdbpp.set_trace()
# x = self.x_basis = ( # assert len(flat._array) == len(self.shm._array[fields])
# np.broadcast_to(
# shm._array['index'][:, None], x = self.x_basis = (
# # self.flat._array.shape, np.broadcast_to(
# self.flat.shape, shm._array['index'][:, None],
# ) + np.array([-0.5, 0, 0, 0.5]) (
shm._array.size,
# 4, # only ohlc
self.flat.shape[1],
),
) + np.array([-0.5, 0, 0, 0.5])
)
# fshm = self.flat = open_shm_array(
# f'{self.name}_flat',
# dtype=flattened.dtype,
# size=flattened.size,
# ) # )
# fshm.push(flattened)
# print(f'unstruct diff: {time.time() - start}')
# profiler('read unstr view bars to line')
# start = self.flat._first.value
ds_curve_r = Renderer( ds_curve_r = Renderer(
flow=self, flow=self,
@ -407,7 +428,7 @@ class Flow(msgspec.Struct): # , frozen=True):
# - if we're **not** downsampling then we simply want to # - if we're **not** downsampling then we simply want to
# render the bars graphics curve and update.. # render the bars graphics curve and update..
# - if insteam we are in a downsamplig state then we to # - if insteam we are in a downsamplig state then we to
x_gt = 8 x_gt = 6
uppx = curve.x_uppx() uppx = curve.x_uppx()
in_line = should_line = curve.isVisible() in_line = should_line = curve.isVisible()
if ( if (
@ -426,37 +447,43 @@ class Flow(msgspec.Struct): # , frozen=True):
# do graphics updates # do graphics updates
if should_line: if should_line:
# start = time.time()
# y = self.shm.unstruct_view(
# ['open', 'high', 'low', 'close'],
# )
# print(f'unstruct diff: {time.time() - start}')
# profiler('read unstr view bars to line')
# # start = self.flat._first.value
# x = self.x_basis[:y.size].flatten() # update flatted ohlc copy
# y = y.flatten() iflat, ishm = self._iflat, self.shm._last.value
# profiler('flattening bars to line') to_update = rfn.structured_to_unstructured(
# path, last = dsc_r.render(read) self.shm._array[iflat:ishm][fields]
# x, flat = ohlc_flat_view( )
# ohlc_shm=self.shm,
# x_basis=x_basis,
# )
# y = y.flatten()
# y_iv = y[ivl:ivr].flatten()
# x_iv = x[ivl:ivr].flatten()
# assert y.size == x.size
x, y = self.flat = ohlc_flatten(array) # print(to_update)
x_iv, y_iv = ohlc_flatten(in_view) self.flat[iflat:ishm][:] = to_update
profiler('flattened OHLC data') profiler('updated ustruct OHLC data')
y_flat = self.flat[:ishm]
x_flat = self.x_basis[:ishm]
self._iflat = ishm
y = y_flat.reshape(-1)
x = x_flat.reshape(-1)
profiler('flattened ustruct OHLC data')
y_iv_flat = y_flat[ivl:ivr]
x_iv_flat = x_flat[ivl:ivr]
y_iv = y_iv_flat.reshape(-1)
x_iv = x_iv_flat.reshape(-1)
profiler('flattened ustruct in-view OHLC data')
# x, y = ohlc_flatten(array)
# x_iv, y_iv = ohlc_flatten(in_view)
# profiler('flattened OHLC data')
curve.update_from_array( curve.update_from_array(
x, x,
y, y,
x_iv=x_iv, x_iv=x_iv,
y_iv=y_iv, y_iv=y_iv,
view_range=None, # hack view_range=(ivl, ivr), # hack
profiler=profiler, profiler=profiler,
) )
profiler('updated ds curve') profiler('updated ds curve')
@ -477,7 +504,7 @@ class Flow(msgspec.Struct): # , frozen=True):
# graphics.setCacheMode(QtWidgets.QGraphicsItem.NoCache) # graphics.setCacheMode(QtWidgets.QGraphicsItem.NoCache)
# graphics.setCacheMode(QtWidgets.QGraphicsItem.DeviceCoordinateCache) # graphics.setCacheMode(QtWidgets.QGraphicsItem.DeviceCoordinateCache)
graphics.prepareGeometryChange() # graphics.prepareGeometryChange()
graphics.update() graphics.update()
if ( if (
@ -632,7 +659,7 @@ class Renderer(msgspec.Struct):
''' '''
# do full source data render to path # do full source data render to path
last_read = ( (
xfirst, xlast, array, xfirst, xlast, array,
ivl, ivr, in_view, ivl, ivr, in_view,
) = self.last_read ) = self.last_read