Flatten out chart tasks

piker/ui/_chart.py

@@ -1,7 +1,7 @@
 """
 High level Qt chart widgets.
 """
-from typing import List, Optional, Tuple
+from typing import List, Optional, Tuple, Dict, Any
 import time
 
 from PyQt5 import QtCore, QtGui
@@ -21,6 +21,9 @@ from ._source import Symbol
 from .. import brokers
 from .. import data
 from ..log import get_logger
+from ._exec import run_qtractor
+from ._source import ohlc_dtype
+from .. import fsp
 
 
 log = get_logger(__name__)
@@ -166,13 +169,13 @@ class LinkedSplitCharts(QtGui.QWidget):
 
         # add crosshairs
         self._ch = CrossHair(
-            parent=self,  #.chart,
+            parent=self,
             # subplots=[plot for plot, d in self.subplots],
             digits=self.digits
         )
         self.chart = self.add_plot(
             name='main',
-            array=array,  #['close'],
+            array=array,
             xaxis=self.xaxis,
             ohlc=True,
         )
@@ -586,40 +589,7 @@ class ChartView(pg.ViewBox):
         self.sigRangeChangedManually.emit(mask)
 
 
-def _main(
+async def add_new_bars(delay_s, linked_charts):
-    sym: str,
-    brokername: str,
-    **qtractor_kwargs,
-) -> None:
-    """Entry point to spawn a chart app.
-    """
-    from ._exec import run_qtractor
-    from ._source import ohlc_dtype
-
-    async def _main(widgets):
-        """Main Qt-trio routine invoked by the Qt loop with
-        the widgets ``dict``.
-        """
-        chart_app = widgets['main']
-
-        # historical data fetch
-        brokermod = brokers.get_brokermod(brokername)
-
-        async with brokermod.get_client() as client:
-            # figure out the exact symbol
-            bars = await client.bars(symbol=sym)
-
-        # remember, msgpack-numpy's ``from_buffer` returns read-only array
-        bars = np.array(bars[list(ohlc_dtype.names)])
-        linked_charts = chart_app.load_symbol(sym, bars)
-
-        # determine ohlc delay between bars
-        times = bars['time']
-
-        # find expected time step between datums
-        delay = times[-1] - times[times != times[-1]][-1]
-
-        async def add_new_bars(delay_s):
     """Task which inserts new bars into the ohlc every ``delay_s`` seconds.
     """
     # TODO: right now we'll spin printing bars if the last time
@@ -657,7 +627,7 @@ def _main(
         new = np.append(
             ohlc,
             np.array(
-                        [(index + 1, t + delay, close, close,
+                [(index + 1, t + delay_s, close, close,
                   close, close, 0)],
                 dtype=ohlc.dtype
             ),
@@ -675,6 +645,39 @@ def _main(
             log.debug("Printing flat line for {sym}")
             linked_charts.update_from_array(ohlc)
 
+
+async def _async_main(
+    sym: str,
+    brokername: str,
+
+    # implicit required argument provided by ``qtractor_run()``
+    widgets: Dict[str, Any],
+
+    # all kwargs are passed through from the CLI entrypoint
+    loglevel: str = None,
+) -> None:
+    """Main Qt-trio routine invoked by the Qt loop with
+    the widgets ``dict``.
+    """
+    chart_app = widgets['main']
+
+    # historical data fetch
+    brokermod = brokers.get_brokermod(brokername)
+
+    async with brokermod.get_client() as client:
+        # figure out the exact symbol
+        bars = await client.bars(symbol=sym)
+
+    # remember, msgpack-numpy's ``from_buffer` returns read-only array
+    bars = np.array(bars[list(ohlc_dtype.names)])
+    linked_charts = chart_app.load_symbol(sym, bars)
+
+    # determine ohlc delay between bars
+    times = bars['time']
+
+    # find expected time step between datums
+    delay = times[-1] - times[times != times[-1]][-1]
+
     async def stream_to_chart(func):
 
         async with tractor.open_nursery() as n:
@@ -697,12 +700,11 @@ def _main(
                         print(tick)
 
     async with trio.open_nursery() as n:
-            from piker import fsp
 
         async with data.open_feed(
             brokername,
             [sym],
-                loglevel=qtractor_kwargs['loglevel'],
+            loglevel=loglevel,
         ) as (fquote, stream):
             # start downstream processor
             n.start_soon(stream_to_chart, fsp.broker_latency)
@@ -711,7 +713,7 @@ def _main(
             quote = await stream.__anext__()
 
             # start graphics tasks after receiving first live quote
-                n.start_soon(add_new_bars, delay)
+            n.start_soon(add_new_bars, delay, linked_charts)
 
             async for quote in stream:
                 ticks = quote.get('ticks')
@@ -722,4 +724,23 @@ def _main(
                                 {'last': tick['price']}
                             )
 
-    run_qtractor(_main, (), ChartSpace, **qtractor_kwargs)
+
+def _main(
+    sym: str,
+    brokername: str,
+    **qtractor_kwargs,
+) -> None:
+    """Sync entry point to start a chart app.
+    """
+    # Qt entry point
+    run_qtractor(
+        # func
+        _async_main,
+        # args,
+        (sym, brokername),
+        # kwargs passed through
+        qtractor_kwargs,
+        # main widget
+        ChartSpace,
+        # **qtractor_kwargs
+    )

piker/ui/_exec.py

@@ -5,6 +5,7 @@ Run ``trio`` in guest mode on top of the Qt event loop.
 All global Qt runtime settings are mostly defined here.
 """
 import traceback
+from typing import Tuple, Callable, Dict
 
 import PyQt5  # noqa
 from pyqtgraph import QtGui
@@ -28,11 +29,11 @@ class MainWindow(QtGui.QMainWindow):
 
 
 def run_qtractor(
-    func,
+    func: Callable,
-    args,
+    args: Tuple,
+    kwargs: Dict,
     main_widget: QtGui.QWidget,
     window_type: QtGui.QMainWindow = MainWindow,
-    loglevel: str = None,
 ) -> None:
     # avoids annoying message when entering debugger from qt loop
     pyqtRemoveInputHook()
@@ -92,10 +93,10 @@ def run_qtractor(
     args = (
         # async_fn
         func,
-        # args
+        # args (always append widgets list)
-        (widgets,),
+        args + (widgets,),
         # kwargs
-        {'loglevel': loglevel},
+        kwargs,
         # arbiter_addr
         (
             tractor._default_arbiter_host,