piker/data/feed.py

@@ -1037,11 +1037,12 @@ async def allocate_persistent_feed(
 
     flume = Flume(
         symbol=symbol,
-        first_quote=first_quote,
-        _rt_shm_token=rt_shm.token,
         _hist_shm_token=hist_shm.token,
+        _rt_shm_token=rt_shm.token,
+        first_quote=first_quote,
         izero_hist=izero_hist,
         izero_rt=izero_rt,
+        # throttle_rate=tick_throttle,
     )
 
     # for ambiguous names we simply apply the retreived

piker/ui/_flows.py

@@ -36,7 +36,6 @@ from PyQt5.QtCore import QLineF
 from ..data._sharedmem import (
     ShmArray,
 )
-from ..data.feed import Flume
 from .._profile import (
     pg_profile_enabled,
     # ms_slower_then,
@@ -209,16 +208,13 @@ class Flow(msgspec.Struct):  # , frozen=True):
     '''
     name: str
     plot: pg.PlotItem
-    _shm: ShmArray
-    flume: Flume
     graphics: Curve | BarItems
-
+    _shm: ShmArray
-    # for tracking y-mn/mx for y-axis auto-ranging
     yrange: tuple[float, float] = None
 
     # in some cases a flow may want to change its
-    # graphical "type" or, "form" when downsampling, to
+    # graphical "type" or, "form" when downsampling,
-    # start this is only ever an interpolation line.
+    # normally this is just a plain line.
     ds_graphics: Optional[Curve] = None
 
     is_ohlc: bool = False
@@ -253,9 +249,9 @@ class Flow(msgspec.Struct):  # , frozen=True):
 
     # TODO: remove this and only allow setting through
     # private ``._shm`` attr?
-    # @shm.setter
+    @shm.setter
-    # def shm(self, shm: ShmArray) -> ShmArray:
+    def shm(self, shm: ShmArray) -> ShmArray:
-    #     self._shm = shm
+        self._shm = shm
 
     def maxmin(
         self,
@@ -322,15 +318,9 @@ class Flow(msgspec.Struct):  # , frozen=True):
 
         '''
         vr = self.plot.viewRect()
-        return (
+        return int(vr.left()), int(vr.right())
-            vr.left(),
-            vr.right(),
-        )
 
-    def datums_range(
+    def datums_range(self) -> tuple[
-        self,
-        index_field: str = 'index',
-    ) -> tuple[
         int, int, int, int, int, int
     ]:
         '''
@@ -338,8 +328,6 @@ class Flow(msgspec.Struct):  # , frozen=True):
 
         '''
         l, r = self.view_range()
-        l = round(l)
-        r = round(r)
 
         # TODO: avoid this and have shm passed
         # in earlier.
@@ -360,23 +348,15 @@ class Flow(msgspec.Struct):  # , frozen=True):
     def read(
         self,
         array_field: Optional[str] = None,
-        index_field: str = 'index',
 
     ) -> tuple[
             int, int, np.ndarray,
             int, int, np.ndarray,
     ]:
-        '''
+        # read call
-        Read the underlying shm array buffer and
-        return the data plus indexes for the first
-        and last
-        which has been written to.
-
-        '''
-        # readable data
         array = self.shm.array
 
-        indexes = array[index_field]
+        indexes = array['index']
         ifirst = indexes[0]
         ilast = indexes[-1]
 

piker/ui/_fsp.py

@@ -42,8 +42,6 @@ from ..data._sharedmem import (
     _Token,
     try_read,
 )
-from ..data.feed import Flume
-from ..data._source import Symbol
 from ._chart import (
     ChartPlotWidget,
     LinkedSplits,
@@ -113,7 +111,7 @@ def update_fsp_chart(
 
     # read from last calculated value and update any label
     last_val_sticky = chart.plotItem.getAxis(
-        'right')._stickies.get(graphics_name)
+        'right')._stickies.get(chart.name)
     if last_val_sticky:
         last = last_row[array_key]
         last_val_sticky.update_from_data(-1, last)
@@ -214,7 +212,7 @@ async def open_fsp_actor_cluster(
 async def run_fsp_ui(
 
     linkedsplits: LinkedSplits,
-    flume: Flume,
+    shm: ShmArray,
     started: trio.Event,
     target: Fsp,
     conf: dict[str, dict],
@@ -251,11 +249,9 @@ async def run_fsp_ui(
             else:
                 chart = linkedsplits.subplots[overlay_with]
 
-            shm = flume.rt_shm
             chart.draw_curve(
-                name,
+                name=name,
-                shm,
+                shm=shm,
-                flume,
                 overlay=True,
                 color='default_light',
                 array_key=name,
@@ -265,9 +261,8 @@ async def run_fsp_ui(
         else:
             # create a new sub-chart widget for this fsp
             chart = linkedsplits.add_plot(
-                name,
+                name=name,
-                shm,
+                shm=shm,
-                flume,
 
                 array_key=name,
                 sidepane=sidepane,
@@ -357,9 +352,6 @@ async def run_fsp_ui(
         #         last = time.time()
 
 
-# TODO: maybe this should be our ``Flow`` type since it maps
-# one flume to the next? The machinery for task/actor mgmt should
-# be part of the instantiation API?
 class FspAdmin:
     '''
     Client API for orchestrating FSP actors and displaying
@@ -371,7 +363,7 @@ class FspAdmin:
         tn: trio.Nursery,
         cluster: dict[str, tractor.Portal],
         linked: LinkedSplits,
-        flume: Flume,
+        src_shm: ShmArray,
 
     ) -> None:
         self.tn = tn
@@ -383,11 +375,7 @@ class FspAdmin:
             tuple[tractor.MsgStream, ShmArray]
         ] = {}
         self._flow_registry: dict[_Token, str] = {}
-
+        self.src_shm = src_shm
-        # TODO: make this a `.src_flume` and add
-        # a `dst_flume`?
-        # (=> but then wouldn't this be the most basic `Flow`?)
-        self.flume = flume
 
     def rr_next_portal(self) -> tractor.Portal:
         name, portal = next(self._rr_next_actor)
@@ -400,7 +388,7 @@ class FspAdmin:
         complete: trio.Event,
         started: trio.Event,
         fqsn: str,
-        dst_fsp_flume: Flume,
+        dst_shm: ShmArray,
         conf: dict,
         target: Fsp,
         loglevel: str,
@@ -421,10 +409,9 @@ class FspAdmin:
                 # data feed key
                 fqsn=fqsn,
 
-                # TODO: pass `Flume.to_msg()`s here?
                 # mems
-                src_shm_token=self.flume.rt_shm.token,
+                src_shm_token=self.src_shm.token,
-                dst_shm_token=dst_fsp_flume.rt_shm.token,
+                dst_shm_token=dst_shm.token,
 
                 # target
                 ns_path=ns_path,
@@ -441,14 +428,12 @@ class FspAdmin:
             ctx.open_stream() as stream,
         ):
 
-            dst_fsp_flume.stream: tractor.MsgStream = stream
-
             # register output data
             self._registry[
                 (fqsn, ns_path)
             ] = (
                 stream,
-                dst_fsp_flume.rt_shm,
+                dst_shm,
                 complete
             )
 
@@ -483,9 +468,9 @@ class FspAdmin:
         worker_name: Optional[str] = None,
         loglevel: str = 'info',
 
-    ) -> (Flume, trio.Event):
+    ) -> (ShmArray, trio.Event):
 
-        fqsn = self.flume.symbol.fqsn
+        fqsn = self.linked.symbol.front_fqsn()
 
         # allocate an output shm array
         key, dst_shm, opened = maybe_mk_fsp_shm(
@@ -493,28 +478,8 @@ class FspAdmin:
             target=target,
             readonly=True,
         )
-
-        portal = self.cluster.get(worker_name) or self.rr_next_portal()
-        provider_tag = portal.channel.uid
-
-        symbol = Symbol(
-            key=key,
-            broker_info={
-                provider_tag: {'asset_type': 'fsp'},
-            },
-        )
-        dst_fsp_flume = Flume(
-            symbol=symbol,
-            _rt_shm_token=dst_shm.token,
-            first_quote={},
-
-            # set to 0 presuming for now that we can't load
-            # FSP history (though we should eventually).
-            izero_hist=0,
-            izero_rt=0,
-        )
         self._flow_registry[(
-            self.flume.rt_shm._token,
+            self.src_shm._token,
             target.name
         )] = dst_shm._token
 
@@ -523,6 +488,7 @@ class FspAdmin:
         #         f'Already started FSP `{fqsn}:{func_name}`'
         #     )
 
+        portal = self.cluster.get(worker_name) or self.rr_next_portal()
         complete = trio.Event()
         started = trio.Event()
         self.tn.start_soon(
@@ -531,13 +497,13 @@ class FspAdmin:
             complete,
             started,
             fqsn,
-            dst_fsp_flume,
+            dst_shm,
             conf,
             target,
             loglevel,
         )
 
-        return dst_fsp_flume, started
+        return dst_shm, started
 
     async def open_fsp_chart(
         self,
@@ -549,7 +515,7 @@ class FspAdmin:
 
     ) -> (trio.Event, ChartPlotWidget):
 
-        flume, started = await self.start_engine_task(
+        shm, started = await self.start_engine_task(
             target,
             conf,
             loglevel,
@@ -561,7 +527,7 @@ class FspAdmin:
                 run_fsp_ui,
 
                 self.linked,
-                flume,
+                shm,
                 started,
                 target,
 
@@ -575,7 +541,7 @@ class FspAdmin:
 @acm
 async def open_fsp_admin(
     linked: LinkedSplits,
-    flume: Flume,
+    src_shm: ShmArray,
     **kwargs,
 
 ) -> AsyncGenerator[dict, dict[str, tractor.Portal]]:
@@ -596,7 +562,7 @@ async def open_fsp_admin(
             tn,
             cluster_map,
             linked,
-            flume,
+            src_shm,
         )
         try:
             yield admin
@@ -610,7 +576,7 @@ async def open_fsp_admin(
 async def open_vlm_displays(
 
     linked: LinkedSplits,
-    flume: Flume,
+    ohlcv: ShmArray,
     dvlm: bool = True,
 
     task_status: TaskStatus[ChartPlotWidget] = trio.TASK_STATUS_IGNORED,
@@ -632,8 +598,6 @@ async def open_vlm_displays(
     sig = inspect.signature(flow_rates.func)
     params = sig.parameters
 
-    ohlcv: ShmArray = flume.rt_shm
-
     async with (
         open_fsp_sidepane(
             linked, {
@@ -653,7 +617,7 @@ async def open_vlm_displays(
                 }
             },
         ) as sidepane,
-        open_fsp_admin(linked, flume) as admin,
+        open_fsp_admin(linked, ohlcv) as admin,
     ):
         # TODO: support updates
         # period_field = sidepane.fields['period']
@@ -669,7 +633,6 @@ async def open_vlm_displays(
         chart = linked.add_plot(
             name='volume',
             shm=shm,
-            flume=flume,
 
             array_key='volume',
             sidepane=sidepane,
@@ -682,8 +645,6 @@ async def open_vlm_displays(
             # the curve item internals are pretty convoluted.
             style='step',
         )
-        # back-link the volume chart to trigger y-autoranging
-        # in the ohlc (parent) chart.
         ohlc_chart.view.enable_auto_yrange(
             src_vb=chart.view,
         )
@@ -749,7 +710,7 @@ async def open_vlm_displays(
 
             tasks_ready = []
             # spawn and overlay $ vlm on the same subchart
-            dvlm_flume, started = await admin.start_engine_task(
+            dvlm_shm, started = await admin.start_engine_task(
                 dolla_vlm,
 
                 {  # fsp engine conf
@@ -846,13 +807,9 @@ async def open_vlm_displays(
                     else:
                         color = 'bracket'
 
-                    assert isinstance(shm, ShmArray)
+                    curve, _ = chart.draw_curve(
-                    assert isinstance(flume, Flume)
+                        name=name,
-
+                        shm=shm,
-                    flow = chart.draw_curve(
-                        name,
-                        shm,
-                        flume,
                         array_key=name,
                         overlay=pi,
                         color=color,
@@ -865,20 +822,20 @@ async def open_vlm_displays(
                     # specially store ref to shm for lookup in display loop
                     # since only a placeholder of `None` is entered in
                     # ``.draw_curve()``.
-                    # flow = chart._flows[name]
+                    flow = chart._flows[name]
                     assert flow.plot is pi
 
             chart_curves(
                 fields,
                 dvlm_pi,
-                dvlm_flume.rt_shm,
+                dvlm_shm,
                 step_mode=True,
             )
 
             # spawn flow rates fsp **ONLY AFTER** the 'dolla_vlm' fsp is
             # up since this one depends on it.
 
-            fr_flume, started = await admin.start_engine_task(
+            fr_shm, started = await admin.start_engine_task(
                 flow_rates,
                 {  # fsp engine conf
                     'func_name': 'flow_rates',
@@ -891,7 +848,7 @@ async def open_vlm_displays(
             # chart_curves(
             #     dvlm_rate_fields,
             #     dvlm_pi,
-            #     fr_flume.rt_shm,
+            #     fr_shm,
             # )
 
             # TODO: is there a way to "sync" the dual axes such that only
@@ -939,7 +896,7 @@ async def open_vlm_displays(
             chart_curves(
                 trade_rate_fields,
                 tr_pi,
-                fr_flume.rt_shm,
+                fr_shm,
                 # step_mode=True,
 
                 # dashed line to represent "individual trades" being
@@ -973,7 +930,7 @@ async def open_vlm_displays(
 async def start_fsp_displays(
 
     linked: LinkedSplits,
-    flume: Flume,
+    ohlcv: ShmArray,
     group_status_key: str,
     loglevel: str,
 
@@ -1016,10 +973,7 @@ async def start_fsp_displays(
     async with (
 
         # NOTE: this admin internally opens an actor cluster
-        open_fsp_admin(
+        open_fsp_admin(linked, ohlcv) as admin,
-            linked,
-            flume,
-        ) as admin,
     ):
         statuses = []
         for target, conf in fsp_conf.items():