added loop and latency args

dranspose/cli.py

@@ -201,6 +201,8 @@ def replay(args: argparse.Namespace) -> None:
         args.port,
         keepalive,
         args.nworkers,
+        latency=args.latency,
+        loop=args.loop,
     )
 
 
@@ -306,6 +308,15 @@ def create_parser() -> argparse.ArgumentParser:
         default=2,
         type=int,
     )
+    parser_replay.add_argument(
+        "--latency",
+        help="time in seconds between replay frames",
+        default=0,
+        type=float,
+    )
+    parser_replay.add_argument(
+        "--loop", action="store_true", help="continuously loop over replay data"
+    )
 
     return parser
 

dranspose/replay.py

@@ -9,6 +9,7 @@
 import time
 import traceback
 from typing import ContextManager, Iterator, Any, Optional, IO, Tuple
+import itertools
 
 import cbor2
 import uvicorn
@@ -257,7 +258,8 @@ def replay(
     broadcast_first: bool = True,
     done_event: threading.Event | None = None,
     start_event: threading.Event | None = None,
-    latency: float | None = None,
+    latency: float = 0,
+    loop: bool = False,
 ) -> None:
     if source is not None:
         sourcecls = utils.import_class(source)
@@ -267,6 +269,8 @@ def replay(
         gens = [get_internals(f) for f in zmq_files]
     else:
         gens = []
+    if loop:
+        gens = list(map(itertools.cycle, gens))
 
     workercls = utils.import_class(wclass)
     logger.info("custom worker class %s", workercls)
@@ -347,9 +351,9 @@ def replay(
                         reducer_app.state.parameters,
                         tick,
                     )
-                if latency is not None:
-                    time.sleep(latency)
+                time.sleep(latency)
             except StopIteration:
+                # gens = [get_internals(f) for f in zmq_files]
                 logger.debug("end of replay, calling finish")
                 _finish(workers, reducer, reducer_app.state.parameters)
                 break
@@ -366,4 +370,3 @@ def replay(
             stop_event.wait()
         except KeyboardInterrupt:
             pass
-        logger.info("replay finished")

pyproject.toml

@@ -110,3 +110,8 @@ extend-exclude = '''
   docs/*
 )
 '''
+
+[dependency-groups]
+dev = [
+    "python-lsp-server>=1.14.0",
+]

tests/test_replay.py

@@ -180,6 +180,64 @@ async def test_replay(
     )
 
 
+@pytest.mark.skipif("config.getoption('rust')", reason="rust does not support dumping")
+@pytest.mark.asyncio
+async def test_replay_looping(
+    controller: None,
+    reducer: Callable[[Optional[str]], Awaitable[None]],
+    create_worker: Callable[[WorkerName], Awaitable[Worker]],
+    create_ingester: Callable[[Ingester], Awaitable[Ingester]],
+    stream_eiger: Callable[[zmq.Context[Any], int, int], Coroutine[Any, Any, None]],
+    stream_orca: Callable[[zmq.Context[Any], int, int], Coroutine[Any, Any, None]],
+    stream_small: Callable[[zmq.Context[Any], int, int], Coroutine[Any, Any, None]],
+    tmp_path: Any,
+) -> None:
+    p_eiger, p_prefix, uuid = await dump_data(
+        reducer,
+        create_worker,
+        create_ingester,
+        stream_eiger,
+        stream_orca,
+        stream_small,
+        tmp_path,
+    )
+    # read dump
+
+    par_file = generate_params(tmp_path)
+    stop_event = threading.Event()
+
+    thread = threading.Thread(
+        target=replay,
+        args=(
+            "tests.aux_payloads:TestWorker",
+            "tests.aux_payloads:TestReducer",
+            [p_eiger, f"{p_prefix}orca-ingester-{uuid}.cbors"],
+            None,
+            par_file,
+        ),
+        kwargs={"port": 5010, "stop_event": stop_event},
+    )
+    thread.start()
+    await asyncio.sleep(2)
+    single_run_len_results = 10
+    test_pass = False
+    for _ in range(10):
+        f = h5pyd.File("http://localhost:5010/", "r", timeout=5)
+        logging.info("file %s", list(f.keys()))
+        len_results = len(f.get("results", []))
+        logging.info("Length of results: %s", len_results)
+        if len_results > single_run_len_results:
+            test_pass = True
+            break
+        await asyncio.sleep(1)
+    assert test_pass, "Results never had more than 10 entries"
+    logging.info("shut down server")
+    stop_event.set()
+    thread.join()
+    await asyncio.sleep(0.1)
+    logging.info("thread joined")
+
+
 @pytest.mark.skipif("config.getoption('rust')", reason="rust does not support dumping")
 @pytest.mark.asyncio
 async def test_replay_gzip(