karabo-bridge-serve-run command

docs/aligning_trains.ipynb

@@ -849,7 +849,7 @@
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "xfel",
+   "display_name": "xfel (Python 3.7)",
    "language": "python",
    "name": "xfel"
   },

docs/cli.rst

@@ -44,6 +44,75 @@ Check the structure of an EuXFEL run or HDF5 file:
 If it finds problems with the data, the program will produce a list of them and
 exit with status 1. See :doc:`validation` for details of what it checks.
 
+.. _cmd-serve-run:
+
+``karabo-bridge-serve-run``
+---------------------------
+
+Stream data from a run, by proposal & run number, in the `Karabo bridge
+<https://rtd.xfel.eu/docs/data-analysis-user-documentation/en/latest/online.html#streaming-from-karabo-bridge>`_
+format. See :doc:`streaming` for more information.
+
+.. code-block:: shell
+
+   #                     Proposal run
+   karabo-bridge-serve-run 700000 40 --port 4321 \
+        --include 'SPB_IRDA_JF4M/DET/JNGFR*:daqOutput' \
+        --include '*/MOTOR/*[*Position]'
+
+.. program:: karabo-bridge-serve-run
+
+.. option:: --port <port>
+
+   Either a numeric TCP port, e.g. ``4321``, or a ZMQ endpoint address, e.g.
+   ``tcp://0.0.0.0:4321``. You will need to give the receiving code this port
+   number or address as well.
+
+   If no port is specified, the program will pick an unused port, and display
+   the endpoint address as it starts.
+
+.. option:: --include <pattern>
+
+   Sources matching the pattern will be included in the streamed data.
+   You can also match keys by putting a key pattern in ``[]`` square brackets
+   at the end of the source pattern.
+
+   You must specify at least one ``--include`` pattern, and you can use the
+   option several times to expand the selection.
+
+   If data is flowing slower than you expect, see if you can use more specific
+   patterns to avoid sending unnecessary sources.
+
+.. option:: --allow-partial
+
+   By default, trains are only sent if they contain all the data selected by
+   ``--include``. This option also sends trains where some of that data is
+   missing.
+
+.. option:: --append-detector-modules
+
+   If the file data contains multiple detector modules as separate sources,
+   i. e. for big area detectors (AGIPD, LPD and DSSC), append these into one
+   single source.
+
+.. option:: --dummy-timestamps
+
+   Add mock timestamps if missing in the original meta-data.
+
+These two options above - appended module sources and dummy timestamps - are
+required if streamed data shall be provided to OnDA.
+
+.. option:: -z <type>, --socket-type <type>
+
+   The ZMQ socket type to use, one of ``PUB``, ``PUSH`` or ``REP``.
+   Default: ``REP``.
+
+.. option:: --use-infiniband
+
+   Use the infiniband network interface (``ib0``) if it's present.
+   This is ignored if ``--port`` is used with a full ZMQ endpoint address.
+
+
 .. _cmd-serve-files:
 
 ``karabo-bridge-serve-files``
@@ -53,6 +122,9 @@ Stream data from files in the `Karabo bridge
 <https://rtd.xfel.eu/docs/data-analysis-user-documentation/en/latest/online.html#streaming-from-karabo-bridge>`_
 format. See :doc:`streaming` for more information.
 
+For streaming data from a run directory, we recommend the newer
+:ref:`cmd-serve-run` command in place of this.
+
 .. code-block:: shell
 
    karabo-bridge-serve-files /gpfs/exfel/exp/XMPL/201750/p700000/proc/r0005 4321

docs/conf.py

@@ -76,7 +76,7 @@
 #
 # This is also used if you do content translation via gettext catalogs.
 # Usually you set "language" from the command line for these cases.
-language = None
+language = 'en'
 
 # There are two options for replacing |today|: either, you set today to some
 # non-false value, then it is used:

docs/streaming.rst

@@ -6,14 +6,20 @@ socket. The ``extra_data`` Python package can stream data from files using the s
 protocol. You can use this to test code which expects to receive data from
 Karabo Bridge, or use the same code for analysing live data and stored data.
 
-To stream the data from a file or run unmodified, use the command::
+To stream data from a saved run, use the ``karabo-bridge-serve-run`` command:
 
-    karabo-bridge-serve-files /gpfs/exfel/exp/SPB/201830/p900022/raw/r0034 4545
+.. code-block:: shell
 
-The number (4545) must be an unused TCP port above 1024. It will bind to
-this and stream the data to any connected clients.
+   #                     Proposal run
+   karabo-bridge-serve-run 700000 40 --port 4545 \
+        --include 'SPB_IRDA_JF4M/DET/JNGFR*:daqOutput' \
+        --include '*/MOTOR/*[*Position]'
+
+The port number (4545 above) must be an unused TCP port above 1024.
+Clients will need this port and the IP address of the sender to connect.
+For clients running on the same node, use the IP address ``127.0.0.1``.
 Command-line options are explained on the
-:ref:`command reference <cmd-serve-files>` page.
+:ref:`command reference <cmd-serve-run>` page.
 
 .. note::
 

extra_data/cli/serve_run.py

@@ -0,0 +1,90 @@
+from argparse import ArgumentParser
+import sys
+
+from .. import open_run
+
+IMPORT_FAILED_MSG = """\
+{}
+
+karabo-bridge-serve-run requires additional dependencies:
+    pip install karabo-bridge psutil
+"""
+
+def main(argv=None):
+    ap = ArgumentParser(prog="karabo-bridge-serve-run")
+    ap.add_argument("proposal", help="Proposal number")
+    ap.add_argument("run", help="Run number")
+    ap.add_argument(
+        "--port", default="0", help="TCP port or ZMQ endpoint to send data on. "
+                                    "Selects a random TCP port by default.")
+    ap.add_argument(
+        "--include", help="Select matching sources (and optionally keys) to "
+                          "include in streamed data",
+        action='append'
+    )
+    ap.add_argument(
+        "--allow-partial", help="Send trains where some sources are missing",
+        action='store_true'
+    )
+    ap.add_argument(
+        "--append-detector-modules", help="combine multiple module sources"
+        " into one (will only work for AGIPD data currently).",
+        action='store_true'
+    )
+    ap.add_argument(
+        "--dummy-timestamps", help="create dummy timestamps if the meta-data"
+        " lacks proper timestamps",
+        action='store_true'
+    )
+    ap.add_argument(
+        "--use-infiniband", help="Use infiniband interface if available "
+                                 "(if a TCP port is specified)",
+        action='store_true'
+    )
+    ap.add_argument(
+        "-z", "--socket-type", help="ZeroMQ socket type",
+        choices=['PUB', 'PUSH', 'REP'], default='REP'
+    )
+    args = ap.parse_args(argv)
+
+    try:
+        from ..export import serve_data
+    except ImportError as e:
+        sys.exit(IMPORT_FAILED_MSG.format(e))
+
+    run = open_run(args.proposal, args.run, data='all')
+
+    if not args.include:
+        print("Available sources:")
+        for s in sorted(run.all_sources):
+            print(f"  {s}")
+        sys.exit("Please select at least one source with --include")
+
+    include = []
+    for pat in args.include:
+        if '[' in pat:
+            if not pat.endswith(']'):
+                sys.exit(f"Missing final ] in {pat!r}")
+            src_pat, key_pat = pat[:-1].split('[', 1)
+            include.append((src_pat, key_pat))
+        else:
+            # Source pattern only
+            include.append(pat)
+
+    if args.allow_partial:
+        sel = run.select(include, require_any=True)
+    else:
+        sel = run.select(include, require_all=True)
+
+    try:
+        serve_data(
+            sel, args.port,
+            append_detector_modules=args.append_detector_modules,
+            dummy_timestamps=args.dummy_timestamps,
+            use_infiniband=args.use_infiniband, sock=args.socket_type
+        )
+    except KeyboardInterrupt:
+        print('\nStopped.')
+
+if __name__ == '__main__':
+    main()

extra_data/export.py

@@ -11,6 +11,8 @@
 """
 
 import os.path as osp
+import time
+from collections import deque
 from socket import AF_INET
 from warnings import warn
 
@@ -107,9 +109,7 @@
         yield tid, train_data
 
 
-def serve_files(path, port, source_glob='*', key_glob='*',
-                append_detector_modules=False, dummy_timestamps=False,
-                use_infiniband=False, sock='REP'):
+def serve_files(path, port, source_glob='*', key_glob='*', **kwargs):
     """Stream data from files through a TCP socket.
 
     Parameters
@@ -144,15 +144,58 @@
         data = H5File(path)
 
     data = data.select(source_glob, key_glob)
+    serve_data(data, port, **kwargs)
 
+
+def serve_data(data, port, append_detector_modules=False,
+                dummy_timestamps=False, use_infiniband=False, sock='REP'):
+    """Stream data from files through a TCP socket.
+
+    Parameters
+    ----------
+    data: DataCollection
+        The data to be streamed; should already have sources & keys selected.
+    port: str or int
+        A ZMQ endpoint (e.g. 'tcp://*:44444') or a TCP port to bind the socket
+        to. Integers or strings of all digits are treated as port numbers.
+    append_detector_modules: bool
+        Combine multi-module detector data in a single data source (sources for
+        individual modules are removed). The last section of the source name is
+        replaces with 'APPEND', example:
+            'SPB_DET_AGIPD1M-1/DET/#CH0:xtdf' -> 'SPB_DET_AGIPD1M-1/DET/APPEND'
+
+        Supported detectors: AGIPD, DSSC, LPD
+    dummy_timestamps: bool
+        Whether to add mock timestamps if the metadata lacks them.
+    use_infiniband: bool
+        Use infiniband interface if available (if port specifies a TCP port)
+    sock: str
+        socket type - supported: REP, PUB, PUSH (default REP).
+    """
     if isinstance(port, int) or port.isdigit():
         endpt = f'tcp://{find_infiniband_ip() if use_infiniband else "*"}:{port}'
     else:
         endpt = port
+
     sender = Sender(endpt, sock=sock, dummy_timestamps=dummy_timestamps)
     print(f'Streamer started on: {sender.endpoint}')
+    ntrains = len(data.train_ids)
+
+    sent_times = deque([time.monotonic()], 10)
+    count = 0
+    tid, rate = 0, 0.
+    def print_update(end='\r'):
+        print(f'Sent {count}/{ntrains} trains - Train ID {tid} - {rate:.1f} Hz', end=end)
+
     for tid, data in _iter_trains(data, merge_detector=append_detector_modules):
         sender.send(data)
+        count += 1
+        new_time = time.monotonic()
+        if count % 5 == 0:
+            rate = len(sent_times) / (new_time - sent_times[0])
+            print_update()
+        sent_times.append(new_time)
+    print_update(end='\n')
 
     # The karabo-bridge code sets linger to 0 so that it doesn't get stuck if
     # the client goes away. But this would also mean that we close the socket

extra_data/read_machinery.py

@@ -4,6 +4,7 @@
 """
 import logging
 import math
+import os
 import os.path as osp
 import re
 import time
@@ -17,7 +18,7 @@
 DETECTOR_NAMES = {'AGIPD', 'DSSC', 'LPD'}
 DETECTOR_SOURCE_RE = re.compile(r'(.+)/DET/(\d+)CH')
 
-DATA_ROOT_DIR = '/gpfs/exfel/exp'
+DATA_ROOT_DIR = os.environ.get('EXTRA_DATA_DATA_ROOT', '/gpfs/exfel/exp')
 
 
 class _SliceConstructor(type):

extra_data/tests/conftest.py

@@ -140,7 +140,7 @@ def mock_spb_raw_run(format_version):
         yield td
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture()
 def mock_spb_raw_and_proc_run():
     with TemporaryDirectory() as td:
         prop_dir = osp.join(str(td), 'SPB', '201830', 'p002012')