Merge pull request #490 from pikers/log_linearized_curve_overlays

Log linearized curve overlays

Author: goodboy

piker/data/_pathops.py

@@ -295,7 +295,7 @@ def slice_from_time(
     arr: np.ndarray,
     start_t: float,
     stop_t: float,
-    step: int | None = None,
+    step: float,  # sampler period step-diff
 
 ) -> slice:
     '''
@@ -324,12 +324,6 @@ def slice_from_time(
     # end of the input array.
     read_i_max = arr.shape[0]
 
-    # TODO: require this is always passed in?
-    if step is None:
-        step = round(t_last - times[-2])
-        if step == 0:
-            step = 1
-
     # compute (presumed) uniform-time-step index offsets
     i_start_t = floor(start_t)
     read_i_start = floor(((i_start_t - t_first) // step)) - 1
@@ -395,7 +389,7 @@ def slice_from_time(
             #     f'diff: {t_diff}\n'
             #     f'REMAPPED START i: {read_i_start} -> {new_read_i_start}\n'
             # )
-            read_i_start = new_read_i_start - 1
+            read_i_start = new_read_i_start
 
     t_iv_stop = times[read_i_stop - 1]
     if (
@@ -412,7 +406,7 @@ def slice_from_time(
             times[read_i_start:],
             # times,
             i_stop_t,
-            side='left',
+            side='right',
         )
 
         if (

piker/data/_sampling.py

@@ -87,7 +87,6 @@ class Sampler:
     # holds all the ``tractor.Context`` remote subscriptions for
     # a particular sample period increment event: all subscribers are
     # notified on a step.
-    # subscribers: dict[int, list[tractor.MsgStream]] = {}
     subscribers: defaultdict[
         float,
         list[
@@ -240,8 +239,11 @@ async def broadcast(
         subscribers for a given sample period.
 
         '''
+        pair: list[float, set]
         pair = self.subscribers[period_s]
 
+        last_ts: float
+        subs: set
         last_ts, subs = pair
 
         task = trio.lowlevel.current_task()
@@ -253,25 +255,35 @@ async def broadcast(
             # f'consumers: {subs}'
         )
         borked: set[tractor.MsgStream] = set()
-        for stream in subs:
+        sent: set[tractor.MsgStream] = set()
+        while True:
             try:
-                await stream.send({
-                    'index': time_stamp or last_ts,
-                    'period': period_s,
-                })
-            except (
-                trio.BrokenResourceError,
-                trio.ClosedResourceError
-            ):
-                log.error(
-                    f'{stream._ctx.chan.uid} dropped connection'
-                )
-                borked.add(stream)
+                for stream in (subs - sent):
+                    try:
+                        await stream.send({
+                            'index': time_stamp or last_ts,
+                            'period': period_s,
+                        })
+                        sent.add(stream)
+
+                    except (
+                        trio.BrokenResourceError,
+                        trio.ClosedResourceError
+                    ):
+                        log.error(
+                            f'{stream._ctx.chan.uid} dropped connection'
+                        )
+                        borked.add(stream)
+                else:
+                    break
+            except RuntimeError:
+                log.warning(f'Client subs {subs} changed while broadcasting')
+                continue
 
         for stream in borked:
             try:
                 subs.remove(stream)
-            except ValueError:
+            except KeyError:
                 log.warning(
                     f'{stream._ctx.chan.uid} sub already removed!?'
                 )
@@ -419,7 +431,7 @@ async def maybe_open_samplerd(
     loglevel: str | None = None,
     **kwargs,
 
-) -> tractor._portal.Portal:  # noqa
+) -> tractor.Portal:  # noqa
     '''
     Client-side helper to maybe startup the ``samplerd`` service
     under the ``pikerd`` tree.
@@ -609,6 +621,14 @@ async def sample_and_broadcast(
             fqsn = f'{broker_symbol}.{brokername}'
             lags: int = 0
 
+            # TODO: speed up this loop in an AOT compiled lang (like
+            # rust or nim or zig) and/or instead of doing a fan out to
+            # TCP sockets here, we add a shm-style tick queue which
+            # readers can pull from instead of placing the burden of
+            # broadcast on solely on this `brokerd` actor. see issues:
+            # - https://github.com/pikers/piker/issues/98
+            # - https://github.com/pikers/piker/issues/107
+
             for (stream, tick_throttle) in subs.copy():
                 try:
                     with trio.move_on_after(0.2) as cs:
@@ -738,9 +758,6 @@ def frame_ticks(
             ticks_by_type[ttype].append(tick)
 
 
-# TODO: a less naive throttler, here's some snippets:
-# token bucket by njs:
-# https://gist.github.com/njsmith/7ea44ec07e901cb78ebe1dd8dd846cb9
 async def uniform_rate_send(
 
     rate: float,
@@ -750,8 +767,22 @@ async def uniform_rate_send(
     task_status: TaskStatus = trio.TASK_STATUS_IGNORED,
 
 ) -> None:
+    '''
+    Throttle a real-time (presumably tick event) stream to a uniform
+    transmissiom rate, normally for the purposes of throttling a data
+    flow being consumed by a graphics rendering actor which itself is limited
+    by a fixed maximum display rate.
 
-    # try not to error-out on overruns of the subscribed (chart) client
+    Though this function isn't documented (nor was intentially written
+    to be) a token-bucket style algo, it effectively operates as one (we
+    think?).
+
+    TODO: a less naive throttler, here's some snippets:
+    token bucket by njs:
+    https://gist.github.com/njsmith/7ea44ec07e901cb78ebe1dd8dd846cb9
+
+    '''
+    # try not to error-out on overruns of the subscribed client
     stream._ctx._backpressure = True
 
     # TODO: compute the approx overhead latency per cycle
@@ -848,6 +879,16 @@ async def uniform_rate_send(
         # rate timing exactly lul
         try:
             await stream.send({sym: first_quote})
+        except tractor.RemoteActorError as rme:
+            if rme.type is not tractor._exceptions.StreamOverrun:
+                raise
+            ctx = stream._ctx
+            chan = ctx.chan
+            log.warning(
+                'Throttled quote-stream overrun!\n'
+                f'{sym}:{ctx.cid}@{chan.uid}'
+            )
+
         except (
             # NOTE: any of these can be raised by ``tractor``'s IPC
             # transport-layer and we want to be highly resilient

piker/data/feed.py

@@ -1589,6 +1589,9 @@ async def open_feed(
                 (brokermod, bfqsns),
             ) in zip(ctxs, providers.items()):
 
+                # NOTE: do it asap to avoid overruns during multi-feed setup?
+                ctx._backpressure = backpressure
+
                 for fqsn, flume_msg in flumes_msg_dict.items():
                     flume = Flume.from_msg(flume_msg)
                     assert flume.symbol.fqsn == fqsn

piker/ui/_annotate.py

@@ -18,7 +18,7 @@
 Annotations for ur faces.
 
 """
-from typing import Callable, Optional
+from typing import Callable
 
 from PyQt5 import QtCore, QtGui, QtWidgets
 from PyQt5.QtCore import QPointF, QRectF
@@ -105,7 +105,7 @@ def __init__(
         get_level: Callable[..., float],
         size: float = 20,
         keep_in_view: bool = True,
-        on_paint: Optional[Callable] = None,
+        on_paint: Callable | None = None,
 
     ) -> None:
 

piker/ui/_axes.py

@@ -1,5 +1,5 @@
 # piker: trading gear for hackers
-# Copyright (C) Tyler Goodlet (in stewardship for piker0)
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
 
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as published by
@@ -20,7 +20,7 @@
 """
 from __future__ import annotations
 from functools import lru_cache
-from typing import Optional, Callable
+from typing import Callable
 from math import floor
 
 import numpy as np
@@ -60,7 +60,8 @@ def __init__(
             **kwargs
         )
 
-        # XXX: pretty sure this makes things slower
+        # XXX: pretty sure this makes things slower!
+        # no idea why given we only move labels for the most part?
         # self.setCacheMode(QtWidgets.QGraphicsItem.DeviceCoordinateCache)
 
         self.pi = plotitem
@@ -190,7 +191,7 @@ def __init__(
         *args,
         min_tick: int = 2,
         title: str = '',
-        formatter: Optional[Callable[[float], str]] = None,
+        formatter: Callable[[float], str] | None = None,
         **kwargs
 
     ) -> None:
@@ -202,8 +203,8 @@ def __init__(
     def set_title(
         self,
         title: str,
-        view: Optional[ChartView] = None,
-        color: Optional[str] = None,
+        view: ChartView | None = None,
+        color: str | None = None,
 
     ) -> Label:
         '''
@@ -303,8 +304,9 @@ def _indexes_to_timestrs(
         viz = chart._vizs[chart.name]
         shm = viz.shm
         array = shm.array
-        times = array['time']
-        i_0, i_l = times[0], times[-1]
+        ifield = viz.index_field
+        index = array[ifield]
+        i_0, i_l = index[0], index[-1]
 
         # edge cases
         if (
@@ -316,11 +318,13 @@ def _indexes_to_timestrs(
             (indexes[0] > i_0
              and indexes[-1] > i_l)
         ):
+            # print(f"x-label indexes empty edge case: {indexes}")
             return []
 
-        if viz.index_field == 'index':
-            arr_len = times.shape[0]
+        if ifield == 'index':
+            arr_len = index.shape[0]
             first = shm._first.value
+            times = array['time']
             epochs = times[
                 list(
                     map(