Merge pull request #215 from BCDA-APS/214-replay

add replay(): replay previous scan(s)

Author: prjemian

CHANGES.rst

@@ -6,6 +6,8 @@ Change History
 
 :1.1.11:  release *tba* : 
 
+    * `#214 <https://github.com/BCDA-APS/apstools/issues/214>`_
+       new: ``apstools.utils.APS_utils.replay()``
     * `#213 <https://github.com/BCDA-APS/apstools/issues/213>`_
        ``list_recent_scans`` show ``exit_status`` 
     * `#212 <https://github.com/BCDA-APS/apstools/issues/212>`_
@@ -38,7 +40,7 @@ Change History
     * `#196 <https://github.com/BCDA-APS/apstools/issues/196>`_
        `spec2ophyd` handle MOTPAR:read_misc_1
     * `#194 <https://github.com/BCDA-APS/apstools/issues/194>`_
-       show table of global ophyd `Signal`s and `Device`s
+       new ``show_ophyd_symbols`` shows table of global ophyd `Signal`s and `Device`s
     * `#193 <https://github.com/BCDA-APS/apstools/issues/193>`_
        `spec2ophyd` ignore None items in SPEC config file
     * `#192 <https://github.com/BCDA-APS/apstools/issues/192>`_
@@ -52,20 +54,20 @@ Change History
 
     * `DEPRECATION <https://github.com/BCDA-APS/apstools/issues/90#issuecomment-483405890>`_
 	   `apstools.plans.run_blocker_in_plan()` will be removed by 2019-12-31.
-	   `Do not write blocking code in bluesky plans.
+	   Do not write blocking code in bluesky plans.
     * Dropped python 3.5 from supported versions
     * `#175 <https://github.com/BCDA-APS/apstools/issues/175>`_
        move `plans.run_in_thread()` to `utils.run_in_thread()`
     * `#168 <https://github.com/BCDA-APS/apstools/issues/168>`_
-       add module to migrate SPEC config file to ophyd setup
+       new `spec2ophyd`  migrates SPEC config file to ophyd setup
     * `#166 <https://github.com/BCDA-APS/apstools/issues/166>`_
        `device_read2table()`: format `device.read()` results in a pyRestTable.Table
     * `#161 <https://github.com/BCDA-APS/apstools/issues/161>`_
        `addDeviceDataAsStream()`: add Device as named document stream event
     * `#159 <https://github.com/BCDA-APS/apstools/issues/159>`_
        convert xlrd.XLRDError into apstools.utils.ExcelReadError
     * `#158 <https://github.com/BCDA-APS/apstools/issues/158>`_
-       run a command list from text file or Excel spreadsheet
+       new ``run_command_file()`` runs a command list from text file or Excel spreadsheet
 
 :1.1.6:  released *2019-05-26*
 

apstools/migration/spec2ophyd.py

@@ -4,6 +4,8 @@
 read SPEC config file and convert to ophyd setup commands
 
 output of ophyd configuration to stdout
+
+*new in apstools release 1.1.7*
 """
 
 

apstools/plans.py

@@ -127,6 +127,8 @@ def execute_command_list(filename, commands, md={}):
         ~summarize_command_file
         ~parse_Excel_command_file
         ~parse_text_command_file
+
+    *new in apstools release 1.1.7*
     """
     full_filename = os.path.abspath(filename)
 
@@ -179,6 +181,7 @@ def get_command_list(filename):
         ~parse_Excel_command_file
         ~parse_text_command_file
 
+    *new in apstools release 1.1.7*
     """
     full_filename = os.path.abspath(filename)
     if not os.path.exists(full_filename):
@@ -376,7 +379,7 @@ def parse_Excel_command_file(filename):
         ~summarize_command_file
         ~parse_text_command_file
 
-
+    *new in apstools release 1.1.7*
     """
     full_filename = os.path.abspath(filename)
     assert os.path.exists(full_filename)
@@ -457,6 +460,7 @@ def parse_text_command_file(filename):
         ~summarize_command_file
         ~parse_Excel_command_file
 
+    *new in apstools release 1.1.7*
     """
     full_filename = os.path.abspath(filename)
     assert os.path.exists(full_filename)
@@ -504,6 +508,7 @@ def register_command_handler(handler=None):
         ~parse_Excel_command_file
         ~parse_text_command_file
 
+    *new in apstools release 1.1.7*
     """
     global _COMMAND_HANDLER_
     _COMMAND_HANDLER_ = handler or execute_command_list
@@ -527,6 +532,7 @@ def run_command_file(filename, md={}):
         ~parse_Excel_command_file
         ~parse_text_command_file
 
+    *new in apstools release 1.1.7*
     """
     commands = get_command_list(filename)
     yield from _COMMAND_HANDLER_(filename, commands)
@@ -617,6 +623,7 @@ def summarize_command_file(filename):
         ~parse_Excel_command_file
         ~parse_text_command_file
 
+    *new in apstools release 1.1.7*
     """
     commands = get_command_list(filename)
     print(f"Command file: {filename}")

apstools/utils.py

@@ -20,6 +20,7 @@
    ~pairwise
    ~print_snapshot_list
    ~print_RE_md
+   ~replay
    ~run_in_thread
    ~show_ophyd_symbols
    ~split_quoted_line
@@ -40,7 +41,9 @@
 # The full license is in the file LICENSE.txt, distributed with this software.
 #-----------------------------------------------------------------------------
 
+from bluesky.callbacks.best_effort import BestEffortCallback
 from collections import OrderedDict
+import databroker
 import datetime
 from email.mime.text import MIMEText
 from event_model import NumpyEncoder
@@ -190,10 +193,12 @@ def list_recent_scans(num=20, keys=[], printing=True, show_command=False, db=Non
 
     PARAMETERS
     
-    num : int (default: ``20``)
+    num : int
         Make the table include the ``num`` most recent scans.
-    keys : [str] (default: ``[]``)
+        (default: ``20``)
+    keys : [str]
         Include these additional keys from the start document.
+        (default: ``[]``)
         
         Two special keys are supported:
         
@@ -203,13 +208,16 @@ def list_recent_scans(num=20, keys=[], printing=True, show_command=False, db=Non
           Also, it will be truncated so that it is no more than 40 characters.)
         * ``exit_status`` : from the stop document
 
-    printing : bool (default: ``True``)
+    printing : bool
         If True, print the table to stdout
-    show_command : bool (default: ``False``)
+        (default: ``True``)
+    show_command : bool
         If True, show the (reconstructed) full command,
         but truncate it to no more than 40 characters)
-    db : object (default: ``db`` from the IPython shell)
+        (default: ``False``)
+    db : object
         Instance of ``databroker.Broker()``
+        (default: ``db`` from the IPython shell)
 
     RETURNS
     
@@ -229,13 +237,9 @@ def list_recent_scans(num=20, keys=[], printing=True, show_command=False, db=Non
         f17f026   2019-07-25 16:19:04.929030 149     count     testing     4845 
         ========= ========================== ======= ========= =========== =====
 
+    *new in apstools release 1.1.10*
     """
-    try:
-        from IPython import get_ipython
-        global_db = get_ipython().user_ns["db"]
-    except AttributeError as _exc:
-        global_db = None
-    db = db or global_db
+    db = db or ipython_shell_namespace()["db"]
 
     if show_command:
         labels = "scan_id  command".split() + keys
@@ -299,12 +303,7 @@ def print_RE_md(dictionary=None, fmt="simple", printing=True):
         ======================== ===================================
     
     """
-    try:
-        from IPython import get_ipython
-        RE = get_ipython().user_ns["RE"]
-    except AttributeError as _exc:
-        RE = None
-    dictionary = dictionary or RE.md
+    dictionary = dictionary or ipython_shell_namespace()["RE"].md
     md = dict(dictionary)   # copy of input for editing
     v = dictionary_table(md["versions"], fmt=fmt)   # sub-table
     md["versions"] = str(v).rstrip()
@@ -335,6 +334,42 @@ def pairwise(iterable):
     return zip(a, a)
 
 
+def replay(headers, callback=None):
+    """
+    replay the document stream from one (or more) scans (headers)
+    
+    PARAMETERS
+    
+    headers: scan or [scan]
+        Scan(s) to be replayed through callback.
+        A *scan* is an instance of a Bluesky `databroker.Header`.
+        see: https://nsls-ii.github.io/databroker/api.html?highlight=header#header-api
+    
+    callback: scan or [scan]
+        The Bluesky callback to handle the stream of documents from a scan.
+        If `None`, then use the `bec` (BestEffortCallback) from the IPython shell.
+        (default:`None`)
+
+    *new in apstools release 1.1.11*
+    """
+    callback = callback or ipython_shell_namespace().get(
+        "bec",                  # get from IPython shell
+        BestEffortCallback(),   # make one, if we must
+        )
+    if isinstance(headers, databroker.Header):
+        headers = tuple(headers)
+    for h in headers:
+        if not isinstance(h, databroker.Header):
+            emsg = f"Must be a databroker Header: received: {type(h)}: |{h}|"
+            raise TypeError(emsg)
+        cmd = _rebuild_scan_command(h.start)
+        logger.debug(f"{cmd}")
+        
+        # at last, this is where the real action happens
+        for k, doc in h.documents():    # get the stream
+            callback(k, doc)            # play it through the callback
+
+
 def run_in_thread(func):
     """
     (decorator) run ``func`` in thread
@@ -364,15 +399,19 @@ def show_ophyd_symbols(show_pv=True, printing=True, verbose=False, symbols=None)
     
     PARAMETERS
     
-    show_pv: bool (default: True)
+    show_pv: bool
         If True, also show relevant EPICS PV, if available.
-    printing: bool (default: True)
+        (default: True)
+    printing: bool
         If True, print table to stdout.
-    verbose: bool (default: False)
+        (default: True)
+    verbose: bool
         If True, also show ``str(obj``.
-    symbols: dict (default: `globals()`)
+        (default: False)
+    symbols: dict
         If None, use global symbol table.
         If not None, use provided dictionary. 
+        (default: `globals()`)
     
     RETURNS
     
@@ -402,6 +441,8 @@ def show_ophyd_symbols(show_pv=True, printing=True, verbose=False, symbols=None)
         Out[1]: <pyRestTable.rest_table.Table at 0x7fa4398c7cf8>
         
         In [2]:    
+
+    *new in apstools release 1.1.8*
     """
     table = pyRestTable.Table()
     table.labels = ["name", "ophyd structure"]
@@ -410,12 +451,14 @@ def show_ophyd_symbols(show_pv=True, printing=True, verbose=False, symbols=None)
     if verbose:
         table.addLabel("object representation")
     table.addLabel("label(s)")
-    try:
-        from IPython import get_ipython
-        g = get_ipython().user_ns
-    except AttributeError as _exc:
-        g = globals()
-    g = symbols or g
+    if symbols is None:
+        # the default choice
+        g = ipython_shell_namespace()
+        if len(g) == 0:
+            # ultimate fallback
+            g = globals()
+    else:
+        g = symbols
     for k, v in sorted(g.items()):
         if isinstance(v, (ophyd.Signal, ophyd.Device)):
             row = [k, v.__class__.__name__]
@@ -846,6 +889,18 @@ def ipython_profile_name():
     return get_ipython().profile
 
 
+def ipython_shell_namespace():
+    """
+    get the IPython shell's namespace dictionary (or empty if not found)
+    """
+    try:
+        from IPython import get_ipython
+        ns = get_ipython().user_ns
+    except AttributeError as _exc:
+        ns = {}
+    return ns
+
+
 def print_snapshot_list(db, **search_criteria):
     """
     print (stdout) a list of all snapshots in the databroker

tests/test_utils.py

@@ -172,29 +172,59 @@ def test_show_ophyd_symbols(self):
             if k not in wont_show:
                 self.assertTrue(k in rr, msg)
         self.assertEqual(num, len(table.rows))
-    
-    def test_list_recent_scans(self):
+
+
+class Test_With_Database(unittest.TestCase):
+
+    def setUp(self):
         from tests.test_export_json import get_db
-        db = get_db()
-        headers = db(plan_name="count")
+        self.db = get_db()
+
+    def tearDown(self):
+        pass
+
+    def test_list_recent_scans(self):
+        headers = self.db(plan_name="count")
         headers = list(headers)[0:1]
         self.assertEqual(len(headers), 1)
         table = APS_utils.list_recent_scans(
             keys=["exit_status",], 
             show_command=True,
             printing=False,
             num=10,
-            db=db,
+            db=self.db,
             )
         self.assertIsNotNone(table)
-        self.assertEqual(len(table.labels), 5, "asked for 2 extra columns (total 5)")
-        self.assertEqual(len(table.rows), 10, "asked for 10 rows")
-        self.assertLessEqual(len(table.rows[1][3]), 40, "command row should be 40 char or less")
+        self.assertEqual(
+            len(table.labels), 
+            5, 
+            "asked for 2 extra columns (total 5)")
+        self.assertEqual(
+            len(table.rows), 
+            10, 
+            "asked for 10 rows")
+        self.assertLessEqual(
+            len(table.rows[1][3]), 
+            40, 
+            "command row should be 40 char or less")
+
+    def test_replay(self):
+        replies = []
+        def my_cb(key, doc):
+            replies.append((key, len(doc)))
+        APS_utils.replay(self.db(plan_name="count"), callback=my_cb)
+
+        self.assertGreater(len(replies), 0)
+        keys = set([v[0] for v in replies])
+        for item in "start stop event descriptor datum resource".split():
+            msg = f"{item} not in {keys}"
+            self.assertIn(item, keys, msg)
 
 
 def suite(*args, **kw):
     test_list = [
         Test_Utils,
+        Test_With_Database,
         ]
     test_suite = unittest.TestSuite()
     for test_case in test_list: