From 05497edf9c3598d7448232a4bba2b167c36ad752 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Fri, 14 Aug 2026 15:56:21 +0200 Subject: [PATCH 1/6] Add StreamingActiveLearner: learner loop driven by streamed data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Digital-twin architectures need a learner that is retriggered by streamed input (sensor windows, events) rather than one that drives itself with an internal iteration count and its own simulation task. The existing learners run once to convergence; embedding them in a DT runtime means data arriving after startup never influences learning. StreamingActiveLearner turns the loop data-driven: items arrive via feed() or attached async-iterator sources (pumps deferred to start() so construction needs no event loop), a window of batch_size items (or a max_wait flush, with optional latest-wins conflation) triggers one training -> active-learning -> criterion iteration, and the window is passed to the training task with the previous active-learn result as a warm-start dependency. The stop criterion becomes a publish gate: on_model_ready callbacks fire when it is met, but consumption continues until stop() or source exhaustion — a learner that keeps improving its model as the stream evolves. Verified: 7 unit tests (windowing, publish gate, stop, max_wait flush, conflation, ctor sources without event loop, validation); full unit suite 126 green. Benchmarked at 88k-349k msg/s machinery throughput (in-process feed, batch 1-100). Co-Authored-By: Claude Fable 5 --- rose/al/__init__.py | 2 + rose/al/streaming_learner.py | 257 +++++++++++++++++++++++++++ tests/unit/test_streaming_learner.py | 164 +++++++++++++++++ 3 files changed, 423 insertions(+) create mode 100644 rose/al/streaming_learner.py create mode 100644 tests/unit/test_streaming_learner.py diff --git a/rose/al/__init__.py b/rose/al/__init__.py index fd2e905d..70576f85 100644 --- a/rose/al/__init__.py +++ b/rose/al/__init__.py @@ -1,8 +1,10 @@ from rose.al.active_learner import ParallelActiveLearner, SequentialActiveLearner from rose.al.selector import AlgorithmSelector +from rose.al.streaming_learner import StreamingActiveLearner __all__ = [ "ParallelActiveLearner", "SequentialActiveLearner", + "StreamingActiveLearner", "AlgorithmSelector", ] diff --git a/rose/al/streaming_learner.py b/rose/al/streaming_learner.py new file mode 100644 index 00000000..1427629b --- /dev/null +++ b/rose/al/streaming_learner.py @@ -0,0 +1,257 @@ +import asyncio +import inspect +from collections.abc import AsyncIterator, Callable +from typing import Any + +from radical.asyncflow import WorkflowEngine + +from ..learner import IterationState, Learner, LearnerConfig + + +class StreamingActiveLearner(Learner): + """Active learner driven by streamed data instead of a simulation task. + + Data items arrive via :meth:`feed` or :meth:`attach_source` and are + collected into windows of ``batch_size`` items (a partial window is + flushed after ``max_wait`` seconds). Each window triggers one learning + iteration: training -> active learning -> criterion. The window is + passed as the first positional argument to the training task, and the + previous iteration's active-learn result is appended as a dependency + (for warm starts). + + Unlike SequentialActiveLearner, a met stop criterion does not end the + loop: it marks the model as publishable (``state.should_stop`` is True + and ``on_model_ready`` callbacks fire) while consumption continues. + The loop ends when :meth:`stop` is called or all attached sources are + exhausted. + + Example:: + + learner = StreamingActiveLearner(asyncflow, batch_size=10) + learner.attach_source(sensor_stream()) + + async for state in learner.start(): + if state.should_stop: + publish(state) + """ + + _END = object() # sentinel: an attached source finished + _WAKE = object() # sentinel: stop() unblocking the collector + + def __init__( + self, + asyncflow: WorkflowEngine, + batch_size: int = 1, + max_wait: float | None = None, + conflate: bool = False, + sources: AsyncIterator[Any] | list[AsyncIterator[Any]] | None = None, + ) -> None: + """Initialize the Streaming Active Learner. + + Args: + asyncflow: The workflow engine instance used to manage async tasks. + batch_size: Number of streamed items per learning window. + max_wait: Flush a partial window after this many seconds of + waiting for more items. None waits for a full window. + conflate: If True, drop backlog and keep only the newest + ``batch_size`` items when iterations are slower than the + stream ("latest wins"). + sources: One or more async iterators to consume as data + sources; equivalent to calling :meth:`attach_source` for + each. + """ + super().__init__(asyncflow, register_and_submit=True) + self.batch_size = batch_size + self.max_wait = max_wait + self.conflate = conflate + + self._queue: asyncio.Queue[Any] = asyncio.Queue() + self._open_sources = 0 + self._exhausted = False + self._started = False + self._sources: list[asyncio.Task] = [] + self._pending_sources: list[AsyncIterator[Any]] = [] + self._model_callbacks: list[Callable[[IterationState], Any]] = [] + self._pending_config: LearnerConfig | None = None + + if sources is not None: + for source in sources if isinstance(sources, list) else [sources]: + self.attach_source(source) + + async def feed(self, item: Any) -> None: + """Feed a single data item into the learner's stream.""" + await self._queue.put(item) + + def attach_source(self, source: AsyncIterator[Any]) -> None: + """Attach an async iterator as a data source. + + Sources attached before :meth:`start` are only consumed once the + learner loop runs. The loop ends once all attached sources are + exhausted and the queue is drained; learners fed only via + :meth:`feed` run until :meth:`stop` is called. + """ + self._open_sources += 1 + if self._started: + self._start_pump(source) + else: + self._pending_sources.append(source) + + def _start_pump(self, source: AsyncIterator[Any]) -> None: + async def pump() -> None: + try: + async for item in source: + await self._queue.put(item) + finally: + await self._queue.put(self._END) + + self._sources.append(asyncio.ensure_future(pump())) + + def on_model_ready(self, callback: Callable[[IterationState], Any]) -> None: + """Register a callback fired whenever the stop criterion is met. + + In streaming mode the criterion acts as a publish gate, not a + terminal condition. Callbacks receive the IterationState and may + be sync or async. + """ + self._model_callbacks.append(callback) + + def set_next_config(self, config: LearnerConfig) -> None: + """Set configuration to apply from the next window on.""" + self._pending_config = config + + def stop(self) -> None: + """Signal the learner to stop and unblock the window collector.""" + super().stop() + self._queue.put_nowait(self._WAKE) + + def _drain_sentinel(self, item: Any) -> bool: + """Process a sentinel item; return True if it was one.""" + if item is self._WAKE: + return True + if item is self._END: + self._open_sources -= 1 + if self._open_sources <= 0: + self._exhausted = True + return True + return False + + async def _collect(self) -> list[Any]: + """Collect the next window; empty means stopped or exhausted.""" + window: list[Any] = [] + while len(window) < self.batch_size: + if self.is_stopped or (self._exhausted and self._queue.empty()): + break + try: + timeout = self.max_wait if window else None + item = await asyncio.wait_for(self._queue.get(), timeout) + except asyncio.TimeoutError: + break # flush partial window + if not self._drain_sentinel(item): + window.append(item) + + if self.conflate: + while not self._queue.empty(): + item = self._queue.get_nowait() + if not self._drain_sentinel(item): + window.append(item) + window = window[-self.batch_size :] + + return window + + async def start( + self, initial_config: LearnerConfig | None = None + ) -> AsyncIterator[IterationState]: + """Consume the stream and yield an IterationState per window. + + Args: + initial_config: Optional LearnerConfig; can be replaced between + windows via set_next_config(). + + Yields: + IterationState per processed window, with ``window_size`` in + its state dict. ``should_stop`` marks criterion-met (model + ready) states; the loop itself keeps running. + """ + if not self.training_function or not self.active_learn_function: + raise ValueError("Training and Active Learning functions must be set!") + + self._started = True + for source in self._pending_sources: + self._start_pump(source) + self._pending_sources.clear() + + config = initial_config + acl_task: Any = None + _stop_reason = "stream_exhausted" + + try: + i = 0 + while True: + window = await self._collect() + if self.is_stopped: + _stop_reason = "stopped" + break + if not window: + break # sources exhausted + + if self._pending_config is not None: + config = self._pending_config + self._pending_config = None + + self.clear_state() + + train_cfg = self._get_iteration_task_config( + self.training_function, config, "training", i + ) + train_cfg["args"] = (window, *train_cfg["args"]) + train_task = self._register_task(train_cfg, deps=acl_task) + train_result = await train_task + + acl_cfg = self._get_iteration_task_config( + self.active_learn_function, config, "active_learn", i + ) + acl_task = self._register_task(acl_cfg, deps=train_task) + acl_result = await acl_task + + if self.is_stopped: + _stop_reason = "stopped" + break + self._extract_state_from_result(train_result) + self._extract_state_from_result(acl_result) + + metric_value: float | None = None + should_stop = False + if self.criterion_function: + crit_cfg = self._get_iteration_task_config( + self.criterion_function, config, "criterion", i + ) + stop_result = await self._register_task(crit_cfg) + if self.is_stopped: + _stop_reason = "stopped" + break + should_stop, metric_value = self._check_stop_criterion(stop_result) + + self.register_state("window_size", len(window)) + self._iteration_state = self.build_iteration_state( + iteration=i, + metric_value=metric_value, + should_stop=should_stop, + current_config=config, + ) + + self._notify_trackers_iteration(self._iteration_state) + if should_stop: + for cb in self._model_callbacks: + result = cb(self._iteration_state) + if inspect.isawaitable(result): + await result + + yield self._iteration_state + i += 1 + except Exception: + _stop_reason = "error" + raise + finally: + self._notify_trackers_stop(self._iteration_state, _stop_reason) + for task in self._sources: + task.cancel() diff --git a/tests/unit/test_streaming_learner.py b/tests/unit/test_streaming_learner.py new file mode 100644 index 00000000..9fcc1176 --- /dev/null +++ b/tests/unit/test_streaming_learner.py @@ -0,0 +1,164 @@ +"""Unit tests for StreamingActiveLearner: windowing, publish-gate criterion, +stop() and source-exhaustion termination.""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock + +import pytest +from radical.asyncflow import WorkflowEngine + +from rose.al.streaming_learner import StreamingActiveLearner + + +def make_learner(batch_size=2, max_wait=None, conflate=False, criterion_results=None, sources=None): + """Create a learner with mocked task submission. + + Training results echo the received window; criterion results are taken + from the given list (metric, compared as '< 1.0'). + """ + learner = StreamingActiveLearner( + MagicMock(spec=WorkflowEngine), + batch_size=batch_size, + max_wait=max_wait, + conflate=conflate, + sources=sources, + ) + learner.training_function = {"func": AsyncMock(), "args": (), "kwargs": {}, "decor_kwargs": {}} + learner.active_learn_function = { + "func": AsyncMock(), + "args": (), + "kwargs": {}, + "decor_kwargs": {}, + } + metrics = iter(criterion_results or ()) + if criterion_results is not None: + learner.criterion_function = { + "func": AsyncMock(), + "args": (), + "kwargs": {}, + "decor_kwargs": {}, + "operator": "<", + "threshold": 1.0, + "metric_name": "test_metric", + } + + windows = [] + + async def mock_reg(task_obj, deps=None): + if task_obj.get("metric_name"): + return next(metrics) + if task_obj["args"]: # training task: window is first arg + windows.append(task_obj["args"][0]) + return "result" + + learner._register_task = AsyncMock(side_effect=mock_reg) + return learner, windows + + +@pytest.mark.asyncio +async def test_windows_batched_and_source_exhaustion(): + learner, windows = make_learner(batch_size=2) + + async def source(): + for i in range(4): + yield i + + learner.attach_source(source()) + + states = [state async for state in learner.start()] + + assert windows == [[0, 1], [2, 3]] + assert [s.iteration for s in states] == [0, 1] + assert all(s.window_size == 2 for s in states) + + +def test_sources_at_construction_need_no_event_loop(): + # pumps are deferred to start(), so construction works outside a loop + async def source(): + for i in range(4): + yield i + + learner, windows = make_learner(batch_size=2, sources=source()) + + async def run(): + states = [state async for state in learner.start()] + assert windows == [[0, 1], [2, 3]] + assert len(states) == 2 + + asyncio.run(run()) + + +@pytest.mark.asyncio +async def test_criterion_is_publish_gate_not_termination(): + learner, _ = make_learner(batch_size=1, criterion_results=[0.5, 2.0]) + published = [] + learner.on_model_ready(published.append) + + await learner.feed("a") + await learner.feed("b") + + states = [] + + async def run(): + async for state in learner.start(): + states.append(state) + if len(states) == 2: + learner.stop() + + await asyncio.wait_for(run(), timeout=5.0) + + # criterion met on first window (0.5 < 1.0) but loop continued + assert [s.should_stop for s in states] == [True, False] + assert len(published) == 1 and published[0] is states[0] + + +@pytest.mark.asyncio +async def test_stop_unblocks_empty_queue(): + learner, _ = make_learner() + + async def run(): + async for _ in learner.start(): + pytest.fail("no data was fed, no state expected") + + task = asyncio.ensure_future(run()) + await asyncio.sleep(0.1) + learner.stop() + await asyncio.wait_for(task, timeout=5.0) + assert learner.is_stopped + + +@pytest.mark.asyncio +async def test_max_wait_flushes_partial_window(): + learner, windows = make_learner(batch_size=10, max_wait=0.05) + + await learner.feed(1) + await learner.feed(2) + + async def run(): + async for _ in learner.start(): + learner.stop() + + await asyncio.wait_for(run(), timeout=5.0) + assert windows == [[1, 2]] + + +@pytest.mark.asyncio +async def test_conflate_keeps_newest_items(): + learner, windows = make_learner(batch_size=2, conflate=True) + for i in range(6): + await learner.feed(i) + + async def run(): + async for _ in learner.start(): + learner.stop() + + await asyncio.wait_for(run(), timeout=5.0) + assert windows == [[4, 5]] + + +@pytest.mark.asyncio +async def test_missing_tasks_raise(): + learner = StreamingActiveLearner(MagicMock(spec=WorkflowEngine)) + with pytest.raises(ValueError): + async for _ in learner.start(): + pass From 04307ab0c782ed878a2b77fd8bfebb030ea45d64 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Fri, 14 Aug 2026 16:00:32 +0200 Subject: [PATCH 2/6] Apply docformatter wrapping for pre-commit CI The repo's pre-commit hook (docformatter, wrap at 100) re-wraps docstring paragraphs; run it locally so CI proceeds to the test jobs. Co-Authored-By: Claude Fable 5 --- rose/al/streaming_learner.py | 10 ++++------ tests/unit/test_streaming_learner.py | 8 ++++---- 2 files changed, 8 insertions(+), 10 deletions(-) diff --git a/rose/al/streaming_learner.py b/rose/al/streaming_learner.py index 1427629b..d74a5f15 100644 --- a/rose/al/streaming_learner.py +++ b/rose/al/streaming_learner.py @@ -85,9 +85,8 @@ async def feed(self, item: Any) -> None: def attach_source(self, source: AsyncIterator[Any]) -> None: """Attach an async iterator as a data source. - Sources attached before :meth:`start` are only consumed once the - learner loop runs. The loop ends once all attached sources are - exhausted and the queue is drained; learners fed only via + Sources attached before :meth:`start` are only consumed once the learner loop runs. The loop + ends once all attached sources are exhausted and the queue is drained; learners fed only via :meth:`feed` run until :meth:`stop` is called. """ self._open_sources += 1 @@ -109,9 +108,8 @@ async def pump() -> None: def on_model_ready(self, callback: Callable[[IterationState], Any]) -> None: """Register a callback fired whenever the stop criterion is met. - In streaming mode the criterion acts as a publish gate, not a - terminal condition. Callbacks receive the IterationState and may - be sync or async. + In streaming mode the criterion acts as a publish gate, not a terminal condition. Callbacks + receive the IterationState and may be sync or async. """ self._model_callbacks.append(callback) diff --git a/tests/unit/test_streaming_learner.py b/tests/unit/test_streaming_learner.py index 9fcc1176..9e4b63dd 100644 --- a/tests/unit/test_streaming_learner.py +++ b/tests/unit/test_streaming_learner.py @@ -1,5 +1,5 @@ -"""Unit tests for StreamingActiveLearner: windowing, publish-gate criterion, -stop() and source-exhaustion termination.""" +"""Unit tests for StreamingActiveLearner: windowing, publish-gate criterion, stop() and source- +exhaustion termination.""" import asyncio from unittest.mock import AsyncMock, MagicMock @@ -13,8 +13,8 @@ def make_learner(batch_size=2, max_wait=None, conflate=False, criterion_results=None, sources=None): """Create a learner with mocked task submission. - Training results echo the received window; criterion results are taken - from the given list (metric, compared as '< 1.0'). + Training results echo the received window; criterion results are taken from the given list + (metric, compared as '< 1.0'). """ learner = StreamingActiveLearner( MagicMock(spec=WorkflowEngine), From 64330d9cb43c3e13ca67daf0d8ae84a2ae6c3f17 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Fri, 14 Aug 2026 16:06:13 +0200 Subject: [PATCH 3/6] =?UTF-8?q?streaming:=20address=20Copilot=20review=20?= =?UTF-8?q?=E2=80=94=20validate=20params,=20conflate=20at=20ingestion?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit batch_size < 1 and non-positive max_wait now fail fast with a clear ValueError instead of producing empty windows or an asyncio timeout error at runtime. Conflation moves to ingestion time (feed and source pumps drop the oldest backlog before enqueuing), so the queue stays bounded to ~batch_size items even while a slow iteration runs, matching the documented latest-wins behavior. Co-Authored-By: Claude Fable 5 --- rose/al/streaming_learner.py | 21 +++++++++++++++++---- tests/unit/test_streaming_learner.py | 16 ++++++++++++++++ 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/rose/al/streaming_learner.py b/rose/al/streaming_learner.py index d74a5f15..153f2fe4 100644 --- a/rose/al/streaming_learner.py +++ b/rose/al/streaming_learner.py @@ -53,14 +53,19 @@ def __init__( batch_size: Number of streamed items per learning window. max_wait: Flush a partial window after this many seconds of waiting for more items. None waits for a full window. - conflate: If True, drop backlog and keep only the newest - ``batch_size`` items when iterations are slower than the - stream ("latest wins"). + conflate: If True, drop the oldest backlog at ingestion time and + keep only the newest ~``batch_size`` items when iterations + are slower than the stream ("latest wins"); the internal + queue stays bounded. sources: One or more async iterators to consume as data sources; equivalent to calling :meth:`attach_source` for each. """ super().__init__(asyncflow, register_and_submit=True) + if batch_size < 1: + raise ValueError(f"batch_size must be >= 1, got {batch_size}") + if max_wait is not None and max_wait <= 0: + raise ValueError(f"max_wait must be positive or None, got {max_wait}") self.batch_size = batch_size self.max_wait = max_wait self.conflate = conflate @@ -80,6 +85,14 @@ def __init__( async def feed(self, item: Any) -> None: """Feed a single data item into the learner's stream.""" + await self._put(item) + + async def _put(self, item: Any) -> None: + """Enqueue a data item; when conflating, drop the oldest backlog first so the queue stays + bounded to roughly ``batch_size`` items even while an iteration is running.""" + if self.conflate: + while self._queue.qsize() >= self.batch_size: + self._drain_sentinel(self._queue.get_nowait()) await self._queue.put(item) def attach_source(self, source: AsyncIterator[Any]) -> None: @@ -99,7 +112,7 @@ def _start_pump(self, source: AsyncIterator[Any]) -> None: async def pump() -> None: try: async for item in source: - await self._queue.put(item) + await self._put(item) finally: await self._queue.put(self._END) diff --git a/tests/unit/test_streaming_learner.py b/tests/unit/test_streaming_learner.py index 9e4b63dd..da8eaa5d 100644 --- a/tests/unit/test_streaming_learner.py +++ b/tests/unit/test_streaming_learner.py @@ -162,3 +162,19 @@ async def test_missing_tasks_raise(): with pytest.raises(ValueError): async for _ in learner.start(): pass + + +def test_invalid_params_raise(): + with pytest.raises(ValueError): + StreamingActiveLearner(MagicMock(spec=WorkflowEngine), batch_size=0) + with pytest.raises(ValueError): + StreamingActiveLearner(MagicMock(spec=WorkflowEngine), max_wait=-1.0) + + +@pytest.mark.asyncio +async def test_conflate_bounds_queue_at_ingestion(): + learner, _ = make_learner(batch_size=2, conflate=True) + for i in range(100): + await learner.feed(i) + # backlog is dropped at ingestion, not at window collection + assert learner._queue.qsize() <= 2 From abbe07b72af7c19ed1640866761bfff422f38742 Mon Sep 17 00:00:00 2001 From: Benjamin Carter Date: Fri, 14 Aug 2026 16:20:57 -0400 Subject: [PATCH 4/6] Add draft example of StreamingActiveLearner --- examples/active_learn/streaming/run_me.py | 90 +++++++++++++++++++++++ 1 file changed, 90 insertions(+) create mode 100644 examples/active_learn/streaming/run_me.py diff --git a/examples/active_learn/streaming/run_me.py b/examples/active_learn/streaming/run_me.py new file mode 100644 index 00000000..b0f68272 --- /dev/null +++ b/examples/active_learn/streaming/run_me.py @@ -0,0 +1,90 @@ +# This is a *draft* example of the StreamingActiveLearner +# It cuts some corners in order to deliver a bare-bones test. +# Namely: StreamingActiveLearner currently requires a stop criterion. +# so, I have a dummy one that always returns true. +# (and technically the stop criterion is usually executable, but +# did this for brevity) +# +# All tasks are function tasks. Currently, the only way to fetch state in +# the model is if they are function tasks returning dictionaries. If these were +# executable tasks, the developer would be responsible for their own data transfer +# management. +# +# A more "real" example of the StreamingActiveLearner would eventually replace +# this quick test script. + +import asyncio +from concurrent.futures import ProcessPoolExecutor +import os +import sys + +from radical.asyncflow import WorkflowEngine +from rhapsody.backends import ConcurrentExecutionBackend + +from rose.al.streaming_learner import StreamingActiveLearner +from rose.learner import IterationState +from rose.metrics import MEAN_SQUARED_ERROR_MSE + + +async def rose_al(): + engine = await ConcurrentExecutionBackend(ProcessPoolExecutor()) + asyncflow = await WorkflowEngine.create(engine) + + acl = StreamingActiveLearner(asyncflow, batch_size=2) + + # Define and register the simulation task + @acl.simulation_task(as_executable=False) + async def simulation(window, *args): + print(f"Start simulation: {window}") + return {"train_window": window} + + # Define and register the training task + @acl.training_task(as_executable=False) + async def training(train_window, *args): + print(f"Start training: {train_window}") + out = [] + for i in train_window: + out.append(i * 2) + return out + + # Define and register the active learning task + @acl.active_learn_task(as_executable=False) + async def active_learn(from_train, *args): + print(f"Start learning: {from_train}") + out = [] + for i in from_train: + out.append(i * 3) + return {"sum": sum(out)} + + # Defining the stop criterion with a metric (MSE in this case) + @acl.as_stop_criterion( + as_executable=False, metric_name=MEAN_SQUARED_ERROR_MSE, threshold=0.1 + ) + async def check_mse(*args): + return 0.01 + + # dummy data source + async def data_source(): + for i in range(10): + yield i + await asyncio.sleep(1) + + acl.attach_source(data_source()) + + # model callback + async def on_model_callback(state: IterationState): + print(f"Publish: {state.sum}") + + acl.on_model_ready(on_model_callback) + + # start learner + async for i in acl.start(): + await asyncio.sleep(0) + + acl.stop() + + await acl.shutdown() + + +if __name__ == "__main__": + asyncio.run(rose_al()) From 65d71150b520122332fa7b8d3fb8a38dd8815b37 Mon Sep 17 00:00:00 2001 From: Benjamin Carter Date: Fri, 14 Aug 2026 16:33:31 -0400 Subject: [PATCH 5/6] Style fix for pre-commit hook --- examples/active_learn/streaming/run_me.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/examples/active_learn/streaming/run_me.py b/examples/active_learn/streaming/run_me.py index b0f68272..92c38391 100644 --- a/examples/active_learn/streaming/run_me.py +++ b/examples/active_learn/streaming/run_me.py @@ -15,8 +15,6 @@ import asyncio from concurrent.futures import ProcessPoolExecutor -import os -import sys from radical.asyncflow import WorkflowEngine from rhapsody.backends import ConcurrentExecutionBackend @@ -78,7 +76,7 @@ async def on_model_callback(state: IterationState): acl.on_model_ready(on_model_callback) # start learner - async for i in acl.start(): + async for _ in acl.start(): await asyncio.sleep(0) acl.stop() From 8d0912df8ceebaf76cd8de3e7ccc6bbe1b467d2c Mon Sep 17 00:00:00 2001 From: Benjamin Carter Date: Fri, 14 Aug 2026 16:46:24 -0400 Subject: [PATCH 6/6] Uses ruff vs black --- examples/active_learn/streaming/run_me.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/examples/active_learn/streaming/run_me.py b/examples/active_learn/streaming/run_me.py index 92c38391..51a7498f 100644 --- a/examples/active_learn/streaming/run_me.py +++ b/examples/active_learn/streaming/run_me.py @@ -55,9 +55,7 @@ async def active_learn(from_train, *args): return {"sum": sum(out)} # Defining the stop criterion with a metric (MSE in this case) - @acl.as_stop_criterion( - as_executable=False, metric_name=MEAN_SQUARED_ERROR_MSE, threshold=0.1 - ) + @acl.as_stop_criterion(as_executable=False, metric_name=MEAN_SQUARED_ERROR_MSE, threshold=0.1) async def check_mse(*args): return 0.01