Overlays interaction latency tuning

piker/_daemon.py

@@ -257,7 +257,7 @@ async def open_piker_runtime(
     # and spawn the service tree distributed per that.
     start_method: str = 'trio',
 
-    tractor_kwargs: dict = {},
+    **tractor_kwargs,
 
 ) -> tuple[
     tractor.Actor,

piker/_profile.py

@@ -152,9 +152,14 @@ def __new__(
             # don't do anything
             return cls._disabledProfiler
 
-        # create an actual profiling object
         cls._depth += 1
         obj = super(Profiler, cls).__new__(cls)
+        obj._msgs = []
+
+        # create an actual profiling object
+        if cls._depth < 1:
+            cls._msgs = []
+
         obj._name = msg or func_qualname
         obj._delayed = delayed
         obj._markCount = 0
@@ -174,8 +179,12 @@ def __call__(self, msg=None):
 
         self._markCount += 1
         newTime = perf_counter()
+        tot_ms = (newTime - self._firstTime) * 1000
         ms = (newTime - self._lastTime) * 1000
-        self._newMsg("  %s: %0.4f ms", msg, ms)
+        self._newMsg(
+            f"  {msg}: {ms:0.4f}, tot:{tot_ms:0.4f}"
+        )
+
         self._lastTime = newTime
 
     def mark(self, msg=None):

piker/data/_formatters.py

@@ -55,6 +55,10 @@ class IncrementalFormatter(msgspec.Struct):
     shm: ShmArray
     viz: Viz
 
+    # the value to be multiplied any any index into the x/y_1d arrays
+    # given the input index is based on the original source data array.
+    flat_index_ratio: float = 1
+
     @property
     def index_field(self) -> 'str':
         '''
@@ -92,8 +96,8 @@ def xy_slice(self) -> slice:
     xy_nd_stop: int | None = None
 
     # TODO: eventually incrementally update 1d-pre-graphics path data?
-    # x_1d: Optional[np.ndarray] = None
-    # y_1d: Optional[np.ndarray] = None
+    x_1d: np.ndarray | None = None
+    y_1d: np.ndarray | None = None
 
     # incremental view-change state(s) tracking
     _last_vr: tuple[float, float] | None = None
@@ -107,32 +111,6 @@ def index_step_size(self) -> float:
         '''
         return self.viz.index_step()
 
-    def __repr__(self) -> str:
-        msg = (
-            f'{type(self)}: ->\n\n'
-            f'fqsn={self.viz.name}\n'
-            f'shm_name={self.shm.token["shm_name"]}\n\n'
-
-            f'last_vr={self._last_vr}\n'
-            f'last_ivdr={self._last_ivdr}\n\n'
-
-            f'xy_slice={self.xy_slice}\n'
-            # f'xy_nd_stop={self.xy_nd_stop}\n\n'
-        )
-
-        x_nd_len = 0
-        y_nd_len = 0
-        if self.x_nd is not None:
-            x_nd_len = len(self.x_nd)
-            y_nd_len = len(self.y_nd)
-
-        msg += (
-            f'x_nd_len={x_nd_len}\n'
-            f'y_nd_len={y_nd_len}\n'
-        )
-
-        return msg
-
     def diff(
         self,
         new_read: tuple[np.ndarray],
@@ -180,8 +158,6 @@ def diff(
             # set us in a zero-to-append state
             nd_stop = self.xy_nd_stop = src_stop
 
-            align_index = array[self.index_field]
-
         # compute the length diffs between the first/last index entry in
         # the input data and the last indexes we have on record from the
         # last time we updated the curve index.
@@ -334,6 +310,9 @@ def format_to_1d(
             array = in_view
             profiler(f'{self.viz.name} view range slice {view_range}')
 
+        # TODO: we need to check if the last-datum-in-view is true and
+        # if so only slice to the 2nd last datumonly slice to the 2nd
+        # last datum.
         # hist = array[:slice_to_head]
 
         # XXX: WOA WTF TRACTOR DEBUGGING BUGGG
@@ -353,6 +332,11 @@ def format_to_1d(
             array_key,
             view_range,
         )
+        # cache/save last 1d outputs for use by other
+        # readers (eg. `Viz.draw_last_datum()` in the
+        # only-draw-last-uppx case).
+        self.x_1d = x_1d
+        self.y_1d = y_1d
 
         # app_tres = None
         # if append_len:
@@ -376,11 +360,6 @@ def format_to_1d(
         # update the last "in view data range"
         if len(x_1d):
             self._last_ivdr = x_1d[0], x_1d[-1]
-            if (
-                self.index_field == 'time'
-                and (x_1d[-1] == 0.5).any()
-            ):
-                breakpoint()
 
         profiler('.format_to_1d()')
 
@@ -503,14 +482,22 @@ def format_xy_nd_to_1d(
         # NOTE: we don't include the very last datum which is filled in
         # normally by another graphics object.
         x_1d = array[self.index_field][:-1]
-        if (
-            self.index_field == 'time'
-            and x_1d.any()
-            and (x_1d[-1] == 0.5).any()
-        ):
-            breakpoint()
-
         y_1d = array[array_key][:-1]
+
+        # name = self.viz.name
+        # if 'trade_rate' == name:
+        #     s = 4
+        #     x_nd = list(self.x_nd[self.xy_slice][-s:-1])
+        #     y_nd = list(self.y_nd[self.xy_slice][-s:-1])
+        #     print(
+        #         f'{name}:\n'
+        #         f'XY data:\n'
+        #         f'x: {x_nd}\n'
+        #         f'y: {y_nd}\n\n'
+        #         f'x_1d: {list(x_1d[-s:])}\n'
+        #         f'y_1d: {list(y_1d[-s:])}\n\n'
+
+        #     )
         return (
             x_1d,
             y_1d,
@@ -532,6 +519,7 @@ class OHLCBarsFmtr(IncrementalFormatter):
     fields: list[str] = field(
         default_factory=lambda: ['open', 'high', 'low', 'close']
     )
+    flat_index_ratio: float = 4
 
     def allocate_xy_nd(
         self,
@@ -627,7 +615,7 @@ def format_xy_nd_to_1d(
 
         '''
         x, y, c = path_arrays_from_ohlc(
-            array,
+            array[:-1],
             start,
             bar_w=self.index_step_size,
             bar_gap=w * self.index_step_size,
@@ -826,13 +814,6 @@ def format_xy_nd_to_1d(
         x_1d = x_step_iv.reshape(x_step_iv.size)
         y_1d = y_step_iv.reshape(y_step_iv.size)
 
-        if (
-            self.index_field == 'time'
-            and x_1d.any()
-            and (x_1d == 0.5).any()
-        ):
-            breakpoint()
-
         # debugging
         # if y_1d.any():
         #     s = 6

piker/data/_m4.py

@@ -91,6 +91,14 @@ def ds_m4(
         x_end = x[-1]  # x end value/highest in domain
         xrange = (x_end - x_start)
 
+    if xrange < 0:
+        log.error(f'-VE M4 X-RANGE: {x_start} -> {x_end}')
+        # XXX: broken x-range calc-case, likely the x-end points
+        # are wrong and have some default value set (such as
+        # x_end -> <some epoch float> while x_start -> 0.5).
+        # breakpoint()
+        return None
+
     # XXX: always round up on the input pixels
     # lnx = len(x)
     # uppx *= max(4 / (1 + math.log(uppx, 2)), 1)

piker/data/_pathops.py

@@ -17,6 +17,11 @@
 Super fast ``QPainterPath`` generation related operator routines.
 
 """
+from math import (
+    ceil,
+    floor,
+)
+
 import numpy as np
 from numpy.lib import recfunctions as rfn
 from numba import (
@@ -58,20 +63,27 @@ def xy_downsample(
     # downsample whenever more then 1 pixels per datum can be shown.
     # always refresh data bounds until we get diffing
     # working properly, see above..
-    bins, x, y, ymn, ymx = ds_m4(
+    m4_out = ds_m4(
         x,
         y,
         uppx,
     )
 
-    # flatten output to 1d arrays suitable for path-graphics generation.
-    x = np.broadcast_to(x[:, None], y.shape)
-    x = (x + np.array(
-        [-x_spacer, 0, 0, x_spacer]
-    )).flatten()
-    y = y.flatten()
+    if m4_out is not None:
+        bins, x, y, ymn, ymx = m4_out
+        # flatten output to 1d arrays suitable for path-graphics generation.
+        x = np.broadcast_to(x[:, None], y.shape)
+        x = (x + np.array(
+            [-x_spacer, 0, 0, x_spacer]
+        )).flatten()
+        y = y.flatten()
 
-    return x, y, ymn, ymx
+        return x, y, ymn, ymx
+
+    # XXX: we accept a None output for the case where the input range
+    # to ``ds_m4()`` is bad (-ve) and we want to catch and debug
+    # that (seemingly super rare) circumstance..
+    return None
 
 
 @njit(
@@ -285,10 +297,7 @@ def slice_from_time(
     stop_t: float,
     step: int | None = None,
 
-) -> tuple[
-    slice,
-    slice,
-]:
+) -> slice:
     '''
     Calculate array indices mapped from a time range and return them in
     a slice.
@@ -308,22 +317,32 @@ def slice_from_time(
     )
 
     times = arr['time']
-    t_first = round(times[0])
+    t_first = floor(times[0])
+    t_last = ceil(times[-1])
 
+    # the greatest index we can return which slices to the
+    # end of the input array.
     read_i_max = arr.shape[0]
 
+    # TODO: require this is always passed in?
     if step is None:
-        step = round(times[-1] - times[-2])
+        step = round(t_last - times[-2])
         if step == 0:
-            # XXX: HOW TF is this happening?
             step = 1
 
     # compute (presumed) uniform-time-step index offsets
-    i_start_t = round(start_t)
-    read_i_start = round(((i_start_t - t_first) // step)) - 1
+    i_start_t = floor(start_t)
+    read_i_start = floor(((i_start_t - t_first) // step)) - 1
+
+    i_stop_t = ceil(stop_t)
 
-    i_stop_t = round(stop_t)
-    read_i_stop = round((i_stop_t - t_first) // step) + 1
+    # XXX: edge case -> always set stop index to last in array whenever
+    # the input stop time is detected to be greater then the equiv time
+    # stamp at that last entry.
+    if i_stop_t >= t_last:
+        read_i_stop = read_i_max
+    else:
+        read_i_stop = ceil((i_stop_t - t_first) // step) + 1
 
     # always clip outputs to array support
     # for read start:
@@ -367,7 +386,7 @@ def slice_from_time(
         #   up_to_arith_start = index[:read_i_start]
 
         if (
-            new_read_i_start < read_i_start
+            new_read_i_start <= read_i_start
         ):
             # t_diff = t_iv_start - start_t
             # print(
@@ -391,14 +410,15 @@ def slice_from_time(
         # )
         new_read_i_stop = np.searchsorted(
             times[read_i_start:],
+            # times,
             i_stop_t,
             side='left',
         )
 
         if (
-            new_read_i_stop < read_i_stop
+            new_read_i_stop <= read_i_stop
         ):
-            read_i_stop = read_i_start + new_read_i_stop
+            read_i_stop = read_i_start + new_read_i_stop + 1
 
     # sanity checks for range size
     # samples = (i_stop_t - i_start_t) // step