Factor ohlc to line data conversion into `._pathops.ohlc_to_line()`

piker/ui/_flows.py

@@ -52,13 +52,13 @@ from .._profile import (
 )
 from ._pathops import (
     gen_ohlc_qpath,
+    ohlc_to_line,
 )
 from ._ohlc import (
     BarItems,
 )
 from ._curve import (
     FastAppendCurve,
-    # step_path_arrays_from_1d,
 )
 from ..log import get_logger
 
@@ -426,29 +426,34 @@ class Flow(msgspec.Struct):  # , frozen=True):
                     # create a flattened view onto the OHLC array
                     # which can be read as a line-style format
                     shm = self.shm
-
-                    # flat = self.gy = self.shm.unstruct_view(fields)
-                    self.gy = self.shm.ustruct(fields)
-                    first = self._iflat_first = self.shm._first.value
-                    last = self._iflat_last = self.shm._last.value
-
-                    # write pushed data to flattened copy
-                    self.gy[first:last] = rfn.structured_to_unstructured(
-                        self.shm.array[fields]
-                    )
-
-                    # generate an flat-interpolated x-domain
-                    self.gx = (
-                        np.broadcast_to(
-                            shm._array['index'][:, None],
                     (
-                                shm._array.size,
+                        self._iflat_first,
-                                # 4,  # only ohlc
+                        self._iflat_last,
-                                self.gy.shape[1],
+                        self.gx,
-                            ),
+                        self.gy,
-                        ) + np.array([-0.5, 0, 0, 0.5])
+                    ) = ohlc_to_line(shm)
-                    )
+
-                    assert self.gy.any()
+                    # self.gy = self.shm.ustruct(fields)
+                    # first = self._iflat_first = self.shm._first.value
+                    # last = self._iflat_last = self.shm._last.value
+
+                    # # write pushed data to flattened copy
+                    # self.gy[first:last] = rfn.structured_to_unstructured(
+                    #     self.shm.array[fields]
+                    # )
+
+                    # # generate an flat-interpolated x-domain
+                    # self.gx = (
+                    #     np.broadcast_to(
+                    #         shm._array['index'][:, None],
+                    #         (
+                    #             shm._array.size,
+                    #             # 4,  # only ohlc
+                    #             self.gy.shape[1],
+                    #         ),
+                    #     ) + np.array([-0.5, 0, 0, 0.5])
+                    # )
+                    # assert self.gy.any()
 
                 # print(f'unstruct diff: {time.time() - start}')
                 # profiler('read unstr view bars to line')

piker/ui/_pathops.py

@@ -22,11 +22,15 @@ from typing import (
 )
 
 import numpy as np
+from numpy.lib import recfunctions as rfn
 from numba import njit, float64, int64  # , optional
 import pyqtgraph as pg
 from PyQt5 import QtGui
 # from PyQt5.QtCore import QLineF, QPointF
 
+from ..data._sharedmem import (
+    ShmArray,
+)
 from .._profile import pg_profile_enabled, ms_slower_then
 from ._compression import (
     # ohlc_flatten,
@@ -176,3 +180,49 @@ def gen_ohlc_qpath(
     profiler("generate path with arrayToQPath")
 
     return path
+
+
+def ohlc_to_line(
+    ohlc_shm: ShmArray,
+    fields: list[str] = ['open', 'high', 'low', 'close']
+
+) -> tuple[
+    int,  # flattened first index
+    int,  # flattened last index
+    np.ndarray,
+    np.ndarray,
+]:
+    '''
+    Convert an input struct-array holding OHLC samples into a pair of
+    flattened x, y arrays with the same size (datums wise) as the source
+    data.
+
+    '''
+    y_out = ohlc_shm.ustruct(fields)
+    first = ohlc_shm._first.value
+    last = ohlc_shm._last.value
+
+    # write pushed data to flattened copy
+    y_out[first:last] = rfn.structured_to_unstructured(
+        ohlc_shm.array[fields]
+    )
+
+    # generate an flat-interpolated x-domain
+    x_out = (
+        np.broadcast_to(
+            ohlc_shm._array['index'][:, None],
+            (
+                ohlc_shm._array.size,
+                # 4,  # only ohlc
+                y_out.shape[1],
+            ),
+        ) + np.array([-0.5, 0, 0, 0.5])
+    )
+    assert y_out.any()
+
+    return (
+        first,
+        last,
+        x_out,
+        y_out,
+    )