diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cb7a2c4..bf5b249 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -101,3 +101,42 @@ jobs: run: >- pytest test/integration -rs --continue-on-collection-errors --timeout=120 --timeout-method=thread + api: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + + # The radical dependencies must not come from naive PyPI resolution: + # PyPI's rhapsody-py 0.4.0 lacks `rhapsody.backends.execution.orbit` + # -- the backend `digitaltwin.service` imports -- which was the + # collection-blocking ModuleNotFoundError this workflow documented. + # Pre-installed pinned, so resolving digitaltwin's requirements + # keeps them. asyncflow 0.5.1 carries the non-main-thread fix; + # orbit is floored, not pinned, since 0.7.0 (dispatcher dialect). + - name: Install pinned radical dependencies + run: >- + pip install "radical.asyncflow==0.5.1" "radical.orbit>=0.7" + "rhapsody-py[telemetry] @ git+https://github.com/radical-cybertools/rhapsody@e491cd2" + "rose @ git+https://github.com/radical-cybertools/ROSE@64330d9cb43c3e13ca67daf0d8ae84a2ae6c3f17" + + # src/ layout: pytest tests the *installed* package, never the + # working tree. Always install before testing -- a stale install + # will silently pass against old code. + - name: Install digitaltwin (test + service extras) + run: pip install ".[test,service,learn]" pytest-timeout + + # --continue-on-collection-errors: one broken test module (see the + # PR description) must not hide the rest of the suite behind an + # "Interrupted" abort. --timeout guards against a hanging test + # taking down the whole job; it turns a hang into a failed test + # with a traceback. + - name: Run api tests + run: >- + pytest test/api_test -rs --continue-on-collection-errors + --timeout=120 --timeout-method=thread diff --git a/src/digitaltwin/components.py b/src/digitaltwin/components.py index 04e3892..251d05e 100644 --- a/src/digitaltwin/components.py +++ b/src/digitaltwin/components.py @@ -389,6 +389,7 @@ def __init__(self, name: str, hard: bool = True) -> None: self.count_hard = 0 self.count_soft = 0 self.set_soft = False + self.recv_soft = 0 def __str__(self) -> str: return self.name @@ -423,15 +424,18 @@ async def put(self, in_data: TypedData): dtype = in_data.dtype if not (self.dtypes[dtype]): # soft. just store the result + if len(self.previous[dtype]) == 0: + self.recv_soft += 1 + if self.previous_retain.get(dtype, True): self.previous[dtype] = [in_data.data] self.previous_retain[dtype] = False else: self.previous[dtype].append(in_data.data) - if not self.set_soft: + + if self.count_hard == 0 and not self.set_soft: self.set_soft = True - for _ in range(self.count_soft): - self._update.release() + self._update.release() return def predicate(): @@ -487,7 +491,16 @@ async def run(self): self.condition.notify_all() self.condition.release() - for i in range(self.count_hard + self.count_soft): + # The update will only fire until ALL the soft items gets something. + if self.recv_soft < self.count_soft: + await asyncio.sleep(0.01) + continue + + # did all hard vals update. + for i in range(self.count_hard): + await self._update.acquire() + + if self.count_hard == 0: await self._update.acquire() self.set_soft = False @@ -495,7 +508,6 @@ async def run(self): if self.dtypes[dtype]: continue # emit on any soft barriers - # drain previous in reverse append order self.output_queues[dtype].put_nowait( WindowedTypeData( diff --git a/src/digitaltwin/runtime.py b/src/digitaltwin/runtime.py index 8183933..b77ce6a 100644 --- a/src/digitaltwin/runtime.py +++ b/src/digitaltwin/runtime.py @@ -13,6 +13,7 @@ from collections import defaultdict, deque from contextvars import ContextVar from dataclasses import dataclass, field +import sys from typing import cast try: @@ -1018,6 +1019,7 @@ def _record_error(self, exc: BaseException | str): error, exc_info=exc if isinstance(exc, BaseException) else None, ) + print(f"twin component failed: {error}", file=sys.stderr) if self.state is RuntimeState.FAILED: # the cause is already recorded, and its teardown is running @@ -1282,6 +1284,8 @@ async def _internal_agent_inference(self, in_data: TypedData, req_dtype: DataTyp return TypedData(NULL_DTYPE, None) return answer + raise ValueError("No component found") + # add a barrier def add_barrier(self, barrier: Barrier) -> None: """Register a synchronization barrier. diff --git a/test/api_test/README.md b/test/api_test/README.md new file mode 100644 index 0000000..98f1cc3 --- /dev/null +++ b/test/api_test/README.md @@ -0,0 +1,48 @@ +# API test coverage mapping + +This maps the `api_test/conftest.py` checklist of DT framework promises to the +existing numbered demos under `test/` (00-11, 100), as a starting point for +writing real pytests that check the framework does what it promises, not just +that individual functions don't crash. These are not unit tests. + +## Checklist → Demo coverage + +| Checklist item | Covered by | Confidence | +|---|---|---| +| **ADD_INPUT** | 01, 04 (`runtime.add_input` binds external sensor channel) | Solid, but only via manual two-terminal run | +| **ADD_TASK - persistent** | 02, 03, 05, 06, 07, 08, 09, 10, 11 | Solid | +| **ADD_TASK - non-persistent** | Every demo's `data_sink` | Solid but incidental — nothing asserts non-persistence specifically | +| **ADD_INVESTIGATOR - input callback** | 03, 06 (gregory/nilakantha/monte_carlo), 100 | Solid | +| **ADD_INVESTIGATOR - inference task** | Nearly all demos | Solid | +| **ADD_INVESTIGATOR - inference task update** | 02, 03, 04, 05, 06, 100 (`publish_new_model`) | Solid | +| **ADD_AGENT - Model selector task** | 04, 05, 06, 09, 11, 100 | Solid | +| **ADD_AGENT - Model publish task** | 06, 100 (`model_publish_cb` override) | Solid | +| **ADD_AGENT - filter input task** | none | **Zero coverage** — `ON_FILTERED_INPUT`/`ON_FILTERED_OUTPUT` exist in runtime.py but no demo subscribes to them | +| **ADD_AGENT - inter-agent inference** | 100 only (`get_inference` chained through profiler→endpoint) | Solid but entangled with the profiler harness | +| **ADD_AGENT - Model selector update** | 04, 05, 06, 09, 11, 100 | Solid | +| **ADD_AGENT - Multi Investigators** | 05, 06, 11 | Solid | +| **ADD_BARRIER - Hard barrier** | 07 — but the hard-barrier block is **commented out** in `run_me.py` | **Effectively uncovered** | +| **ADD_BARRIER - Soft default barrier** | 07 (a/b/c soft dtypes) | Solid | +| **ADD_BARRIER - Hard(slow)/soft(fast)** | 07 mixes delays but doesn't isolate/assert this pairing | Gap | +| **ADD_BARRIER - Soft(fast)/hard(slow)** | Same — happens incidentally in 07, never asserted | Gap | +| **ADD_DATA_JOIN - Data Join** | 08 (real join), 09 (trivial single-dtype no-op join) | Solid | +| **ADD_DATA_SPLIT - task** | 09, 10 | Solid | +| **ADD_DATA_SPLIT - a None** | none — no demo's split ever returns a fully-None result | **Gap** | +| **ADD_DATA_SPLIT - one None, one Item** | 10 (`HighLow`) | Solid | +| **ADD_DATA_SPLIT - both items** | none — no split ever returns two live `TypedData` in one call | **Gap** | + +## Gaps (zero real coverage today) + +1. **Filter input task** (`ON_FILTERED_INPUT`/`ON_FILTERED_OUTPUT`) — mechanism exists, nothing exercises it. +2. **Hard barrier** — only example is commented out in `07-barrier/run_me.py`. +3. **Hard/soft speed-pairing semantics** — 07 runs a 5-sensor mix but never isolates or asserts either ordering. +4. **Data split → all-None result**. +5. **Data split → both outputs populated simultaneously**. + +## Behaviors seen in demos but not on the checklist + +- Basic lifecycle (start/stop/redeploy progression across 00→01→02) +- Shared sub-tasks across investigators (`11-shared-sim`: `register_shared_subtask`/`get_shared_subtask`/`call_shared_subtask`) +- Remote/distributed orchestration (`09-remote`: `RemoteDTOrchestrator`, `runtime.package`, `register_user_modules`) +- Resource-aware model selection (`100`) — really a richer version of "inter-agent inference" +- Windowed data reads underlying the soft barrier (`WindowDataType`/`WindowedTypeData`) diff --git a/test/api_test/components.py b/test/api_test/components.py new file mode 100644 index 0000000..7f9a78f --- /dev/null +++ b/test/api_test/components.py @@ -0,0 +1,301 @@ +"""Model investigator for the api_test digital twin. + +`StampModel`'s inference task simply stamps every input with the current +model version and a timestamp. The version is bumped every `UPDATE_EVERY` +inputs via the `on_input` callback, which then republishes the model -- +mirroring `03-conditional-redeploy/model.py`'s conditional-redeploy pattern, +except the trigger is an input count instead of a value threshold. +""" + +import asyncio +import random +import string +import time + +from digitaltwin import SplitTask +from radical.asyncflow import WorkflowEngine +from digitaltwin.components import ModelInvestigator, SciAgent, TypedData +from digitaltwin.runtime import RuntimeAPI + +from dtypes import ( + FLIP_AGENT_IN, + FLIP_AGENT_OUT, + INVESTIGATOR_OUT_DTYPE, + AGENT_OUT_DTYPE, + NEG_NUM, + POS_NUM, +) + +UPDATE_EVERY = 2 + + +class InvestigatorTest(ModelInvestigator): + def __init__(self): + super().__init__(None) + self.version = 1 + self.count = 0 + self.to_publish = asyncio.Event() + + async def do_inference(in_data: TypedData, version=1): + return TypedData( + INVESTIGATOR_OUT_DTYPE, + { + "version": version, + "timestamp": time.monotonic(), + "dat": in_data.data, + }, + ) + + self.inference_task = do_inference + + async def on_input(self, in_data: TypedData): + self.count += 1 + if self.count % UPDATE_EVERY == 0: + self.version += 1 + self.to_publish.set() + + async def main_loop(self, runtime: RuntimeAPI): + runtime.set_inference_task(self.inference_task) + runtime.publish_new_model({"version": self.version}) + runtime.subscribe_to_topic(runtime.ON_INPUT, self.on_input) + + while True: + await self.to_publish.wait() + runtime.publish_new_model({"version": self.version}) + self.to_publish.clear() + + +class LetterInvestigator(ModelInvestigator): + """One of `TestAgent`'s two investigators: stamps version + timestamp, + same as `TestInvestigator`, plus a random 6-letter sequence -- so a + model-selection test can tell which investigator answered.""" + + def __init__(self, upper: bool = False): + super().__init__(None) + self.version = 0 + self.alphabet = string.ascii_uppercase if upper else string.ascii_lowercase + + async def do_inference( + in_data: TypedData, + version=0, + fcount=0, + fcount_out=0, + out_count=0, + flip=None, + ): + if flip is None: + flip = {} + + letters = "".join(random.choices(self.alphabet, k=6)) + return TypedData( + AGENT_OUT_DTYPE, + { + "version": version, + "timestamp": time.monotonic(), + "letters": letters, + "dat": in_data.data, + "fcount": fcount, + "fcount_out": fcount_out, + "out_count": out_count, + "flip": flip, + }, + ) + + self.inference_task = do_inference + self.to_publish = asyncio.Event() + self.count = 0 + self.filter_count = 0 + self.filter_out_count = 0 + self.out_count = 0 + self.upper = upper + + async def on_input(self, in_data: TypedData): + self.version += 1 + + async def on_filtered_input(self, in_data: TypedData): + self.filter_count += 1 + self.count += 1 + if self.count % UPDATE_EVERY == 0: + self.to_publish.set() + + async def on_filtered_output(self, in_data: TypedData): + self.filter_out_count += 1 + + async def on_output(self, in_data: TypedData): + self.out_count += 1 + + async def main_loop(self, runtime: RuntimeAPI): + runtime.set_inference_task(self.inference_task) + runtime.publish_new_model({"version": self.version}) + runtime.subscribe_to_topic(runtime.ON_INPUT, self.on_input) + runtime.subscribe_to_topic(runtime.ON_FILTERED_INPUT, self.on_filtered_input) + runtime.subscribe_to_topic(runtime.ON_FILTERED_OUTPUT, self.on_filtered_output) + runtime.subscribe_to_topic(runtime.ON_OUTPUT, self.on_output) + while True: + await self.to_publish.wait() + m_arg = { + "version": self.version, + "fcount": self.filter_count, + "fcount_out": self.filter_out_count, + "out_count": self.out_count, + } + runtime.publish_new_model(m_arg) + self.to_publish.clear() + + +class AgentTest(SciAgent): + """A SciAgent with two `LetterInvestigator`s and a pass-through model + selector, mirroring `05-agent-w-multi-investigators/agent.py`.""" + + def __init__(self): + super().__init__(None) + self.inv_lower = LetterInvestigator(upper=False) + self.inv_upper = LetterInvestigator(upper=True) + + self.model_low = {} + self.model_up = {} + + self.update = asyncio.Event() + + async def model_select(in_data: TypedData, i_id, model_kwargs={}): + return i_id, model_kwargs # default to latest model + + self.model_selector = model_select + + async def model_publish_cb( + self, investigator: ModelInvestigator, model_args: dict, acc_metrics: dict + ): + if investigator == self.inv_upper: + self.model_up = model_args + else: + self.model_low = model_args + self.update.set() + + async def main_loop(self, runtime: RuntimeAPI): + runtime.start_investigator(self.inv_lower) + runtime.start_investigator(self.inv_upper) + runtime.set_model_selection_task(self.model_selector) + + # is overwritten by the model publish cb... UP is called last (default model) + runtime.update_model_selector(i_id=self.inv_upper.get_id()) + + toggle = False + while True: + await self.update.wait() + toggle = not (toggle) + marg = {} + if toggle: + inv = self.inv_upper.get_id() + marg = self.model_up + code = "upper" + else: + inv = self.inv_lower.get_id() + marg = self.model_low + code = "LOWER" + + # will call switchcase on code + val = await runtime.get_inference( + TypedData(FLIP_AGENT_IN, code), FLIP_AGENT_OUT + ) + assert val is not None + marg["flip"] = val.data + + runtime.update_model_selector(i_id=inv, model_kwargs=marg) + self.update.clear() + + +class FlipInvestigator(ModelInvestigator): + def __init__(self): + super().__init__(None) + self.version = 0 + + async def do_inference( + in_data: TypedData, + version=0, + fcount=0, + fcount_out=0, + out_count=0, + ): + return TypedData( + FLIP_AGENT_OUT, + { + "version": version, + "timestamp": time.monotonic(), + "swap": in_data.data.swapcase(), + "fcount": fcount, + "fcount_out": fcount_out, + "out_count": out_count, + }, + ) + + self.inference_task = do_inference + self.to_publish = asyncio.Event() + self.count = 0 + self.filter_count = 0 + self.filter_out_count = 0 + self.out_count = 0 + + async def on_input(self, in_data: TypedData): + self.version += 1 + + async def on_filtered_input(self, in_data: TypedData): + self.filter_count += 1 + self.count += 1 + if self.count % UPDATE_EVERY == 0: + self.to_publish.set() + + async def on_filtered_output(self, in_data: TypedData): + self.filter_out_count += 1 + + async def on_output(self, in_data: TypedData): + self.out_count += 1 + + async def main_loop(self, runtime: RuntimeAPI): + runtime.set_inference_task(self.inference_task) + runtime.publish_new_model({"version": self.version}) + runtime.subscribe_to_topic(runtime.ON_INPUT, self.on_input) + runtime.subscribe_to_topic(runtime.ON_FILTERED_INPUT, self.on_filtered_input) + runtime.subscribe_to_topic(runtime.ON_FILTERED_OUTPUT, self.on_filtered_output) + runtime.subscribe_to_topic(runtime.ON_OUTPUT, self.on_output) + + while True: + await self.to_publish.wait() + m_arg = { + "version": self.version, + "fcount": self.filter_count, + "fcount_out": self.filter_out_count, + "out_count": self.out_count, + } + runtime.publish_new_model(m_arg) + self.to_publish.clear() + + +class FlipAgent(SciAgent): + def __init__(self): + super().__init__(None) + self.flip = FlipInvestigator() + + self.update = asyncio.Event() + + async def model_select(in_data: TypedData, i_id, model_kwargs={}): + return i_id # default to latest model + + self.model_selector = model_select + + async def main_loop(self, runtime: RuntimeAPI): + runtime.start_investigator(self.flip) + + runtime.set_model_selection_task(self.model_selector) + runtime.update_model_selector(i_id=self.flip.get_id()) + + +class SplitTest(SplitTask): + def __init__(self): + super().__init__(None) + + async def main_loop(self, runtime: RuntimeAPI, in_data: TypedData): + # runtime + if in_data.data["sensor"] >= 0: + return TypedData(POS_NUM, in_data.data), None + + return None, TypedData(NEG_NUM, in_data.data) diff --git a/test/api_test/conftest.py b/test/api_test/conftest.py new file mode 100644 index 0000000..c1532e4 --- /dev/null +++ b/test/api_test/conftest.py @@ -0,0 +1,109 @@ +# API Test. Do all the features the DT framework promises to the user actually +# work? + + +# DT has: +# ADD_INPUT - DONE +# ADD_TASK - DONE +# - persistent - DONE +# - non-persistent - DONE +# ADD_INVESTIGATOR +# - input callback - DONE +# - inference task - DONE +# - inference task update - DONE +# +# ADD_AGENT +# - Model selector task - DONE +# - Model publish callback - DONE +# - filter input task - DONE +# - inter-agent inference - DONE +# - Model selector update - DONE +# - Multi Investigators - DONE +# +# ADD_BARRIER +# - Hard barrier +# - Soft default barrier +# - Hard (slow) soft (fast) +# - Soft (fast) hard (slow) +# +# ADD_DATA_JOIN +# - Data Join - DONE +# +# ADD_DATA_SPLIT +# - Data split task +# - a None +# - one None, one Item +# - both items +# +# claude --resume 18e73ba2-602f-4436-8e3f-958f132df1b7 + + +import asyncio +import contextlib + +from digitaltwin import PubSubBackend, PubSubConfig, connect_stream_client +import pytest + +from digitaltwin.streaming import ZMQ_BrokerProcess +from sensors import input_sensor + +NAMESPACE = "api_test" + + +@pytest.fixture +async def broker(): + """An embedded stream broker on a random loopback port.""" + + proc = ZMQ_BrokerProcess() + await proc.start() + try: + await asyncio.sleep(1) + yield proc + finally: + await proc.stop() + + +@pytest.fixture +async def stream_clients(broker): + """Factory for namespaced stream clients on the fixture broker. + + All clients it hands out are closed when the test ends -- a client + left open would be caught by the leak assertions of the next test. + """ + + clients = [] + + async def make(namespace: str = NAMESPACE): + client = await connect_stream_client(namespace, *broker.get_connection_str()) + clients.append(client) + return client + + try: + yield make + finally: + for client in clients: + await client.close() + + +@pytest.fixture +async def no_task_leaks(): + """Assert that the test leaves no asyncio task behind.""" + + before = asyncio.all_tasks() + yield + leaked = {task for task in asyncio.all_tasks() if task not in before} + leaked.discard(asyncio.current_task()) + assert not leaked, f"leaked tasks: {leaked}" + + +@pytest.fixture +async def input_sensor_task(stream_clients, broker, no_task_leaks): + s = broker.get_connection_str() + ps_config = PubSubConfig.resolve(NAMESPACE, *s) + tk = asyncio.create_task(input_sensor(ps_config)) + try: + yield tk + finally: + tk.cancel() + with contextlib.suppress(asyncio.CancelledError): + await tk diff --git a/test/api_test/dtypes.py b/test/api_test/dtypes.py new file mode 100644 index 0000000..989962d --- /dev/null +++ b/test/api_test/dtypes.py @@ -0,0 +1,43 @@ +from digitaltwin.components import DataType, JoinDataType + +# one dtype per persistent sensor in sensors.py +PERSIST_SENSOR_DTYPE = DataType("persist_sensor") +FAST_SENSOR_DTYPE = DataType("fast_sensor") +SLOW_SENSOR_DTYPE = DataType("slow_sensor") +FAST2_SENSOR_DTYPE = DataType("fast2_sensor") +SLOW2_SENSOR_DTYPE = DataType("slow2_sensor") +FAST3_SENSOR_DTYPE = DataType("fast3_sensor") +SLOW3_SENSOR_DTYPE = DataType("slow3_sensor") +FAST4_SENSOR_DTYPE = DataType("fast4_sensor") +SLOW4_SENSOR_DTYPE = DataType("slow4_sensor") + +RAND_SENSOR_DTYPE = DataType("rand_sensor") + +# input channel and dtype +INPUT_CHANNEL = "test_input" +INPUT_SENSOR_DTYPE = DataType("input_sensor_dtype") + + +# monitor dtypes + +POST_PERSIST_SENSOR = DataType("Post-Persist") +POST_INPUT = DataType("Post-Input") + +# model investigator dtypes + +INVESTIGATOR_OUT_DTYPE = DataType("investigator_output") + +# science agent dtypes + +AGENT_OUT_DTYPE = DataType("agent_output") +FLIP_AGENT_IN = DataType("flip_agent_in") +FLIP_AGENT_OUT = DataType("flip_agent_out") + +# Data Join Type + +DATA_JOIN = JoinDataType([INVESTIGATOR_OUT_DTYPE, AGENT_OUT_DTYPE]) + +# DATA SPLIT DTYPES + +POS_NUM = DataType("pos") +NEG_NUM = DataType("neg") diff --git a/test/api_test/expected_graph.json b/test/api_test/expected_graph.json new file mode 100644 index 0000000..b2d0fc6 --- /dev/null +++ b/test/api_test/expected_graph.json @@ -0,0 +1,249 @@ +{ + "namespace": "api_test", + "state": "ready", + "last_error": null, + "inputs": [ + { + "dtype": "input_sensor_dtype", + "channel": "test_input", + "codec": "json" + } + ], + "components": [ + { + "component": "Persist_Sensor", + "kind": "utility", + "input_dtype": "TRUE", + "output_dtype": "persist_sensor", + "is_persistent": true, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "Fast_Sensor", + "kind": "utility", + "input_dtype": "TRUE", + "output_dtype": "fast_sensor", + "is_persistent": true, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "Slow_Sensor", + "kind": "utility", + "input_dtype": "TRUE", + "output_dtype": "slow_sensor", + "is_persistent": true, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "Fast2_Sensor", + "kind": "utility", + "input_dtype": "TRUE", + "output_dtype": "fast2_sensor", + "is_persistent": true, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "Slow2_Sensor", + "kind": "utility", + "input_dtype": "TRUE", + "output_dtype": "slow2_sensor", + "is_persistent": true, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "Fast3_Sensor", + "kind": "utility", + "input_dtype": "TRUE", + "output_dtype": "fast3_sensor", + "is_persistent": true, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "Slow3_Sensor", + "kind": "utility", + "input_dtype": "TRUE", + "output_dtype": "slow3_sensor", + "is_persistent": true, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "Fast4_Sensor", + "kind": "utility", + "input_dtype": "TRUE", + "output_dtype": "fast4_sensor", + "is_persistent": true, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "Slow4_Sensor", + "kind": "utility", + "input_dtype": "TRUE", + "output_dtype": "slow4_sensor", + "is_persistent": true, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "Rand_Sensor", + "kind": "utility", + "input_dtype": "TRUE", + "output_dtype": "rand_sensor", + "is_persistent": true, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "InvestigatorTest", + "kind": "investigator", + "input_dtype": "persist_sensor", + "output_dtype": "investigator_output", + "is_persistent": false, + "model_published": false, + "model_keys": [], + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "FlipAgent", + "kind": "agent", + "input_dtype": "flip_agent_in", + "output_dtype": "flip_agent_out", + "is_persistent": false, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "AgentTest", + "kind": "agent", + "input_dtype": "input_sensor_dtype", + "output_dtype": "agent_output", + "is_persistent": false, + "is_join": false, + "is_split": false, + "split_outputs": [] + }, + { + "component": "_JoinComponent", + "kind": "join", + "input_dtype": "investigator_output", + "output_dtype": "JOIN[investigator_output,agent_output]", + "is_persistent": false, + "is_join": true, + "is_split": false, + "split_outputs": [] + }, + { + "component": "_JoinComponent", + "kind": "join", + "input_dtype": "agent_output", + "output_dtype": "JOIN[investigator_output,agent_output]", + "is_persistent": false, + "is_join": true, + "is_split": false, + "split_outputs": [] + }, + { + "component": "SplitTest", + "kind": "split", + "input_dtype": "rand_sensor", + "output_dtype": "NULL", + "is_persistent": false, + "is_join": false, + "is_split": true, + "split_outputs": [ + "pos", + "neg" + ] + } + ], + "dtypes": [ + "JOIN[investigator_output,agent_output]", + "NULL", + "TRUE", + "agent_output", + "fast2_sensor", + "fast3_sensor", + "fast4_sensor", + "fast_sensor", + "flip_agent_in", + "flip_agent_out", + "input_sensor_dtype", + "investigator_output", + "persist_sensor", + "rand_sensor", + "slow2_sensor", + "slow3_sensor", + "slow4_sensor", + "slow_sensor" + ], + "barriers": { + "fast_sensor": [ + { + "name": "HARD_ONLY", + "hard": true + } + ], + "slow_sensor": [ + { + "name": "HARD_ONLY", + "hard": true + } + ], + "fast2_sensor": [ + { + "name": "FAST SOFT", + "hard": false + } + ], + "slow2_sensor": [ + { + "name": "FAST SOFT", + "hard": true + } + ], + "fast3_sensor": [ + { + "name": "SLOW SOFT", + "hard": true + } + ], + "slow3_sensor": [ + { + "name": "SLOW SOFT", + "hard": false + } + ], + "fast4_sensor": [ + { + "name": "SOFT ONLY", + "hard": false + } + ], + "slow4_sensor": [ + { + "name": "SOFT ONLY", + "hard": false + } + ] + } +} \ No newline at end of file diff --git a/test/api_test/monitor.py b/test/api_test/monitor.py new file mode 100644 index 0000000..70b1f58 --- /dev/null +++ b/test/api_test/monitor.py @@ -0,0 +1,21 @@ +import time + +from digitaltwin import NULL_DTYPE, TypedData, UtilityTask +from digitaltwin.components import DataType + + +class MonitorTask(UtilityTask): + def __init__(self, out_dtype: DataType, mark_time=False): + super().__init__(None) + self.output: list[TypedData] = [] + self.out_dtype = out_dtype + self.mark_time = mark_time + + async def main_loop(self, runtime, in_data): + if self.mark_time: + self.output.append({"data": in_data, "recv_time": time.monotonic()}) + else: + self.output.append(in_data) + if self.out_dtype == NULL_DTYPE: + return + return TypedData(self.out_dtype, in_data.data) diff --git a/test/api_test/sensors.py b/test/api_test/sensors.py new file mode 100644 index 0000000..e93c7bd --- /dev/null +++ b/test/api_test/sensors.py @@ -0,0 +1,233 @@ +"""Persistent utility-task sensors for the api_test digital twin. + +Each sensor below is a `UtilityTask` bound with `is_persistent=True` (see +`runtime.add_task(..., TRUTHY, ..., is_persistent=True)` in `run_me.py`). +Its `main_loop` is a single long-running function, structured just like the +external sensor loop in `01-start-inference-stop/sensor.py` (a bounded +`for` loop that computes a value, publishes it, then sleeps) - the only +difference is it publishes in-process via `runtime.stream_config` instead +of through an external `ChannelPublisher`. + +Every sensor emits the same payload shape: `(value, timestamp)`, where +`value` is `random.random()` and `timestamp` is `time.monotonic()` at the +moment of publication. +""" + +import asyncio +import random +import time + +from radical.asyncflow import WorkflowEngine +from digitaltwin.components import UtilityTask +from digitaltwin.streaming import ChannelPublisher + +from dtypes import ( + FAST4_SENSOR_DTYPE, + PERSIST_SENSOR_DTYPE, + FAST_SENSOR_DTYPE, + SLOW4_SENSOR_DTYPE, + SLOW_SENSOR_DTYPE, + FAST2_SENSOR_DTYPE, + SLOW2_SENSOR_DTYPE, + FAST3_SENSOR_DTYPE, + SLOW3_SENSOR_DTYPE, + RAND_SENSOR_DTYPE, + INPUT_CHANNEL, +) + +# tests look at output once everything finishes. +N_ITERS = 12 + + +async def input_sensor(config): + await asyncio.sleep(2) # to wait for broker to start + publisher = await ChannelPublisher.open(INPUT_CHANNEL, config=config) + try: + for i in range(N_ITERS): + await asyncio.sleep(1) + value = {"sensor": i, "sensor_time": time.monotonic()} + print(f"Input_Sensor val: {i}") + await publisher.publish(value) + + finally: + await publisher.close() + + +class Persist_Sensor(UtilityTask): + def __init__(self, flow: WorkflowEngine, delay: float = 0.5): + super().__init__(flow) + self.delay = delay + + async def main_loop(self, runtime, in_data): + await asyncio.sleep(2) + ps = await runtime.stream_config.connect() + try: + for i in range(N_ITERS): + value = {"sensor": i, "sensor_time": time.monotonic()} + print(f"Persist_Sensor val: {i}") + await ps.publish(PERSIST_SENSOR_DTYPE, value) + await asyncio.sleep(self.delay) + finally: + await ps.close() + + +class Fast_Sensor(UtilityTask): + def __init__(self, flow: WorkflowEngine, delay: float = 0.1): + super().__init__(flow) + self.delay = delay + + async def main_loop(self, runtime, in_data): + await asyncio.sleep(2) + ps = await runtime.stream_config.connect() + try: + for i in range(N_ITERS): + value = {"sensor": i, "sensor_time": time.monotonic()} + # print(f"Fast_Sensor val: {value} - {i}") + await ps.publish(FAST_SENSOR_DTYPE, value) + await asyncio.sleep(self.delay) + finally: + await ps.close() + + +class Slow_Sensor(UtilityTask): + def __init__(self, flow: WorkflowEngine, delay: float = 0.5): + super().__init__(flow) + self.delay = delay + + async def main_loop(self, runtime, in_data): + await asyncio.sleep(2) + ps = await runtime.stream_config.connect() + try: + for i in range(N_ITERS): + value = {"sensor": i, "sensor_time": time.monotonic()} + # print(f"Slow_Sensor val: {value} - {i}") + await ps.publish(SLOW_SENSOR_DTYPE, value) + await asyncio.sleep(self.delay) + finally: + await ps.close() + + +class Fast2_Sensor(UtilityTask): + def __init__(self, flow: WorkflowEngine, delay: float = 0.2): + super().__init__(flow) + self.delay = delay + + async def main_loop(self, runtime, in_data): + await asyncio.sleep(2) + ps = await runtime.stream_config.connect() + try: + for i in range(N_ITERS): + value = {"sensor": i, "sensor_time": time.monotonic()} + # print(f"Fast2_Sensor val: {value} - {i}") + await ps.publish(FAST2_SENSOR_DTYPE, value) + await asyncio.sleep(self.delay) + finally: + await ps.close() + + +class Slow2_Sensor(UtilityTask): + def __init__(self, flow: WorkflowEngine, delay: float = 1.0): + super().__init__(flow) + self.delay = delay + + async def main_loop(self, runtime, in_data): + await asyncio.sleep(2) + ps = await runtime.stream_config.connect() + try: + for i in range(N_ITERS): + value = {"sensor": i, "sensor_time": time.monotonic()} + # print(f"Slow2_Sensor val: {value} - {i}") + await ps.publish(SLOW2_SENSOR_DTYPE, value) + await asyncio.sleep(self.delay) + finally: + await ps.close() + + +class Fast3_Sensor(UtilityTask): + def __init__(self, flow: WorkflowEngine, delay: float = 0.25): + super().__init__(flow) + self.delay = delay + + async def main_loop(self, runtime, in_data): + await asyncio.sleep(2) + ps = await runtime.stream_config.connect() + try: + for i in range(N_ITERS): + value = {"sensor": i, "sensor_time": time.monotonic()} + # print(f"Fast3_Sensor val: {value} - {i}") + await ps.publish(FAST3_SENSOR_DTYPE, value) + await asyncio.sleep(self.delay) + finally: + await ps.close() + + +class Slow3_Sensor(UtilityTask): + def __init__(self, flow: WorkflowEngine, delay: float = 1.5): + super().__init__(flow) + self.delay = delay + + async def main_loop(self, runtime, in_data): + await asyncio.sleep(2) + ps = await runtime.stream_config.connect() + try: + for i in range(N_ITERS): + value = {"sensor": i, "sensor_time": time.monotonic()} + # print(f"Slow3_Sensor val: {value} - {i}") + await ps.publish(SLOW3_SENSOR_DTYPE, value) + await asyncio.sleep(self.delay) + finally: + await ps.close() + + +class Fast4_Sensor(UtilityTask): + def __init__(self, flow: WorkflowEngine, delay: float = 0.5): + super().__init__(flow) + self.delay = delay + + async def main_loop(self, runtime, in_data): + await asyncio.sleep(2) + ps = await runtime.stream_config.connect() + try: + for i in range(N_ITERS): + value = {"sensor": i, "sensor_time": time.monotonic()} + # print(f"Fast4_Sensor val: {value} - {i}") + await ps.publish(FAST4_SENSOR_DTYPE, value) + await asyncio.sleep(self.delay) + finally: + await ps.close() + + +class Slow4_Sensor(UtilityTask): + def __init__(self, flow: WorkflowEngine, delay: float = 1): + super().__init__(flow) + self.delay = delay + + async def main_loop(self, runtime, in_data): + await asyncio.sleep(2) + ps = await runtime.stream_config.connect() + try: + for i in range(N_ITERS): + value = {"sensor": i, "sensor_time": time.monotonic()} + # print(f"Slow4_Sensor val: {value} - {i}") + await ps.publish(SLOW4_SENSOR_DTYPE, value) + await asyncio.sleep(self.delay) + finally: + await ps.close() + + +class Rand_Sensor(UtilityTask): + def __init__(self, flow, delay_range: tuple[float, float] = (0.1, 1.0)): + super().__init__(flow) + self.delay_range = delay_range + + async def main_loop(self, runtime, in_data): + await asyncio.sleep(2) + ps = await runtime.stream_config.connect() + try: + for i in range(N_ITERS): + value = {"sensor": i * ((-1) ** i), "sensor_time": time.monotonic()} + # print(f"Rand_Sensor val: {value} - {i}") + await ps.publish(RAND_SENSOR_DTYPE, value) + await asyncio.sleep(random.uniform(*self.delay_range)) + finally: + await ps.close() diff --git a/test/api_test/test_run.py b/test/api_test/test_run.py new file mode 100644 index 0000000..16ca89b --- /dev/null +++ b/test/api_test/test_run.py @@ -0,0 +1,515 @@ +import asyncio +import json +import os +import time + +from digitaltwin import NULL_DTYPE + +from digitaltwin.runtime import DTRuntime +from digitaltwin.streaming import connect_stream_client +from digitaltwin.components import TRUTHY + +from dtypes import * +from sensors import ( + N_ITERS, + Fast4_Sensor, + Persist_Sensor, + Fast_Sensor, + Slow4_Sensor, + Slow_Sensor, + Fast2_Sensor, + Slow2_Sensor, + Fast3_Sensor, + Slow3_Sensor, + Rand_Sensor, +) +from monitor import MonitorTask +from components import UPDATE_EVERY, AgentTest, FlipAgent, InvestigatorTest, SplitTest +from digitaltwin.components import Barrier + + +import logging + +logger = logging.getLogger(__name__) + +# put it all together +# +# input_sensor channel --> INPUT_SENSOR_DTYPE +# Persist_Sensor, Fast_Sensor, Slow_Sensor, Fast2_Sensor, Slow2_Sensor, +# Fast3_Sensor, Slow3_Sensor, Rand_Sensor --> their own dtypes +# +# Nothing consumes these dtypes yet - this just proves persistence and the +# input binding both work end-to-end. +# +# The input sensor is external: run `python -c "import asyncio; +# from sensors import input_sensor; asyncio.run(input_sensor())"` in its +# own terminal once the broker is up. + + +async def setup(stream_clients): + flow = None + + # create the twin's namespaced stream client + pubsub_client = await stream_clients() + + runtime = DTRuntime(flow, pubsub_client) + + # create the persistent sensor tasks + persist_sensor = Persist_Sensor(flow) + fast_sensor = Fast_Sensor(flow) + slow_sensor = Slow_Sensor(flow) + fast2_sensor = Fast2_Sensor(flow) + slow2_sensor = Slow2_Sensor(flow) + fast3_sensor = Fast3_Sensor(flow) + slow3_sensor = Slow3_Sensor(flow) + fast4_sensor = Fast4_Sensor(flow) + slow4_sensor = Slow4_Sensor(flow) + + rand_sensor = Rand_Sensor(flow) + + # the graph opens at its input edge: bind the external sensor's channel + runtime.add_input(INPUT_SENSOR_DTYPE, INPUT_CHANNEL) + + # persistent utility tasks: driven by TRUTHY, publish on their own dtype + runtime.add_task(persist_sensor, TRUTHY, PERSIST_SENSOR_DTYPE, is_persistent=True) + runtime.add_task(fast_sensor, TRUTHY, FAST_SENSOR_DTYPE, is_persistent=True) + runtime.add_task(slow_sensor, TRUTHY, SLOW_SENSOR_DTYPE, is_persistent=True) + runtime.add_task(fast2_sensor, TRUTHY, FAST2_SENSOR_DTYPE, is_persistent=True) + runtime.add_task(slow2_sensor, TRUTHY, SLOW2_SENSOR_DTYPE, is_persistent=True) + runtime.add_task(fast3_sensor, TRUTHY, FAST3_SENSOR_DTYPE, is_persistent=True) + runtime.add_task(slow3_sensor, TRUTHY, SLOW3_SENSOR_DTYPE, is_persistent=True) + runtime.add_task(fast4_sensor, TRUTHY, FAST4_SENSOR_DTYPE, is_persistent=True) + runtime.add_task(slow4_sensor, TRUTHY, SLOW4_SENSOR_DTYPE, is_persistent=True) + runtime.add_task(rand_sensor, TRUTHY, RAND_SENSOR_DTYPE, is_persistent=True) + + # add investigators + investigator = InvestigatorTest() + runtime.add_investigator(investigator, PERSIST_SENSOR_DTYPE, INVESTIGATOR_OUT_DTYPE) + + # add the science agent: its two LetterInvestigators answer with their + # own AGENT_OUT_DTYPE, kept distinct from TestInvestigator's output. + + flip_agent = FlipAgent() + runtime.add_agent(flip_agent, FLIP_AGENT_IN, FLIP_AGENT_OUT) + + # AgentTest depends on FlipAgent. TODO: Make all agent loops only start after start() + agent = AgentTest() + runtime.add_agent(agent, INPUT_SENSOR_DTYPE, AGENT_OUT_DTYPE) + + # Add barriers + hard_only = Barrier("HARD_ONLY") + hard_only.add_dtype(FAST_SENSOR_DTYPE) + hard_only.add_dtype(SLOW_SENSOR_DTYPE) + + fast_soft = Barrier("FAST SOFT") # tests Windowing + fast2_window = fast_soft.add_dtype(FAST2_SENSOR_DTYPE, hard=False) + fast_soft.add_dtype(SLOW2_SENSOR_DTYPE) + + slow_soft = Barrier("SLOW SOFT") # tests replication + slow_soft.add_dtype(FAST3_SENSOR_DTYPE) + slow3_window = slow_soft.add_dtype(SLOW3_SENSOR_DTYPE, hard=False) + + soft_only = Barrier("SOFT ONLY", hard=False) + fast4_window = soft_only.add_dtype(FAST4_SENSOR_DTYPE) + slow4_window = soft_only.add_dtype(SLOW4_SENSOR_DTYPE) + + # add each to runtime. + runtime.add_barrier(hard_only) + runtime.add_barrier(fast_soft) + runtime.add_barrier(slow_soft) + runtime.add_barrier(soft_only) + + # Add a data join + runtime.add_data_join(DATA_JOIN) + + # add data split + st = SplitTest() + runtime.add_data_split_task(st, RAND_SENSOR_DTYPE, [POS_NUM, NEG_NUM]) + + return runtime, fast2_window, slow3_window, fast4_window, slow4_window + + +async def test_setup(stream_clients, no_task_leaks, input_sensor_task): + runtime, _, _, _, _ = await setup(stream_clients) + + if os.path.exists("expected_graph.json"): + with open("expected_graph.json", "r") as f: + # json.dump(runtime.describe(), f, indent=4) + answer = json.load(f) + else: + with open("test/api_test/expected_graph.json", "r") as f: + # json.dump(runtime.describe(), f, indent=4) + answer = json.load(f) + + assert answer == runtime.describe() + + runtime.start() + + # # let it run + await asyncio.sleep(5) + await runtime.stop() + print("Stopped") + + +async def test_run(stream_clients, no_task_leaks, input_sensor_task): + print("Start test run") + runtime, fast2_w, slow3_w, fast4_w, slow4_w = await setup(stream_clients) + + # monitor tasks for the test + input_monitor = MonitorTask(POST_INPUT) + persist_monitor = MonitorTask(POST_PERSIST_SENSOR) + + out_monitor = MonitorTask(NULL_DTYPE) + + pos_monitor = MonitorTask(NULL_DTYPE) + neg_monitor = MonitorTask(NULL_DTYPE) + + # Outputs + output = { + FAST_SENSOR_DTYPE: MonitorTask(NULL_DTYPE, mark_time=True), + SLOW_SENSOR_DTYPE: MonitorTask(NULL_DTYPE, mark_time=True), + fast2_w: MonitorTask(NULL_DTYPE, mark_time=True), + SLOW2_SENSOR_DTYPE: MonitorTask(NULL_DTYPE, mark_time=True), + FAST3_SENSOR_DTYPE: MonitorTask(NULL_DTYPE, mark_time=True), + slow3_w: MonitorTask(NULL_DTYPE, mark_time=True), + fast4_w: MonitorTask(NULL_DTYPE, mark_time=True), + slow4_w: MonitorTask(NULL_DTYPE, mark_time=True), + } + + for dtype, task in output.items(): + runtime.add_task(task, dtype, NULL_DTYPE) + + start_time = time.monotonic() + + # Monitor input and persist task + runtime.add_task(input_monitor, INPUT_SENSOR_DTYPE, POST_INPUT) + runtime.add_task(persist_monitor, PERSIST_SENSOR_DTYPE, POST_PERSIST_SENSOR) + runtime.add_task(out_monitor, DATA_JOIN, NULL_DTYPE) + runtime.add_task(pos_monitor, POS_NUM, NULL_DTYPE) + runtime.add_task(neg_monitor, NEG_NUM, NULL_DTYPE) + + runtime.start() + + # # let it run + await asyncio.sleep(20) + await runtime.stop() + + # should be done. + stop_time = time.monotonic() + + # Check 1: is input ordered? + + # check output + assert len(input_monitor.output) == N_ITERS + + prev_time = start_time + for entry in input_monitor.output: + # early to late + assert entry.data["sensor_time"] > prev_time + + assert input_monitor.output[-1].data["sensor_time"] < stop_time + + # Check two: + # Are all the events from the persistent task there? + assert len(persist_monitor.output) == N_ITERS + + prev_time = start_time + for entry in persist_monitor.output: + # early to late + assert entry.data["sensor_time"] > prev_time + + assert persist_monitor.output[-1].data["sensor_time"] < stop_time + + # Great... Now, add an investigator and an agent. + + # check output from join. + assert len(out_monitor.output) == N_ITERS + i_counter = 0 + for entry in out_monitor.output: + i_entry = entry.data[0] + a_entry = entry.data[1] + + assert i_entry.dtype == INVESTIGATOR_OUT_DTYPE + assert a_entry.dtype == AGENT_OUT_DTYPE + + # the values should match. + assert i_entry.data["dat"]["sensor"] == a_entry.data["dat"]["sensor"] + + # check investigator and auto update + assert i_entry.data["dat"]["sensor"] == i_counter + + # the next one triggers the version update + assert i_entry.data["version"] == (i_counter // UPDATE_EVERY) + 1 + + # check agent with model updates + investigators + print( + a_entry.data["letters"], + i_counter, + a_entry.data["version"], + a_entry.data["fcount"], + a_entry.data["fcount_out"], + a_entry.data["out_count"], + a_entry.data["flip"].get("swap", ""), + a_entry.data["flip"].get("version", 0), + a_entry.data["flip"].get("fcount", 0), + a_entry.data["flip"].get("fcount_out", 0), + a_entry.data["flip"].get("out_count", 0), + ) + assert a_entry.data["letters"].islower() == (i_counter // UPDATE_EVERY) % 2 + + double_up = UPDATE_EVERY * 2 + + assert a_entry.data["fcount"] == (i_counter // double_up) * UPDATE_EVERY + assert a_entry.data["fcount_out"] == a_entry.data["fcount"] + assert a_entry.data["out_count"] == a_entry.data["version"] + + check = ((i_counter // UPDATE_EVERY) - 1) * UPDATE_EVERY + + if check < 0: + check = 0 + assert a_entry.data["version"] == check + + # check inter-agent inference + + i_counter += 1 + + flip = a_entry.data["flip"] + assert a_entry.data["letters"].isupper() == flip["swap"].isupper() + assert flip["fcount"] == flip["fcount_out"] + assert flip["fcount"] == flip["out_count"] + assert flip["version"] == flip["fcount"] + assert flip["version"] == a_entry.data["fcount"] + + assert len(out_monitor.output) == N_ITERS + + # check barriers + print("Barrier check ===") + + # HARD. + + fast_out = output[FAST_SENSOR_DTYPE].output + slow_out = output[SLOW_SENSOR_DTYPE].output + + assert len(fast_out) == len(slow_out) and len(slow_out) == N_ITERS + sorted_recvs = {} + for fast_item, slow_item in zip(fast_out, slow_out): + print( + fast_item["data"].data["sensor"], + fast_item["data"].data["sensor_time"], + fast_item["recv_time"], + "|", + slow_item["data"].data["sensor"], + slow_item["data"].data["sensor_time"], + slow_item["recv_time"], + ) + + # sort by recv time. + sorted_recvs[fast_item["recv_time"]] = "FAST" + sorted_recvs[slow_item["recv_time"]] = "SLOW" + + assert fast_item["data"].data["sensor"] == slow_item["data"].data["sensor"] + + sorted_recvs = dict(sorted(sorted_recvs.items(), key=lambda item: item[0])) + assert len(sorted_recvs) == 2 * N_ITERS + + saw_slow = 0 + saw_fast = 0 + for i in sorted_recvs.values(): + if saw_slow == 1 and saw_fast == 1: + saw_slow = 0 + saw_fast = 0 + + if i == "SLOW" and saw_slow == 0: + saw_slow = 1 + elif i == "FAST" and saw_fast == 0: + saw_fast = 1 + else: + # Failed order test! + raise ValueError("Failed barrier ordering test!") + + # Now, check fast_soft + + print("FAST SOFT check") + + fast_out = output[fast2_w].output + slow_out = output[SLOW2_SENSOR_DTYPE].output + sorted_recvs = {} + + for fast_item, slow_item in zip(fast_out, slow_out): + # get fast window + print( + fast_item["data"].data, + fast_item["recv_time"], + slow_item["data"].data["sensor"], + slow_item["data"].data["sensor_time"], + slow_item["recv_time"], + ) + + # sort by recv time. + for i in fast_item["data"].data: + val = i["sensor"] + t = i["sensor_time"] + + sorted_recvs[t] = {"val": val, "type": "FAST"} + + val = slow_item["data"].data["sensor"] + sorted_recvs[slow_item["data"].data["sensor_time"]] = { + "val": val, + "type": "SLOW", + } + + sorted_recvs = dict(sorted(sorted_recvs.items(), key=lambda item: item[0])) + + # simulate expected: + + prev = None + saw_fast = [] + counter = 0 + + # first is slow, sneak ahead to fast! + r_vals = list(sorted_recvs.values()) + + if r_vals[0]["type"] == "SLOW": + assert r_vals[1]["type"] == "FAST" + prev = r_vals[1]["val"] + + for i in r_vals: + if i["type"] == "SLOW": + if len(saw_fast) == 0: + saw_fast.append(prev) + + # check slow + slow_val = i["val"] + assert slow_val == slow_out[counter]["data"].data["sensor"] + + # check fast + for idx, s in enumerate(fast_out[counter]["data"].data): + assert s["sensor"] == saw_fast[idx] + + # clear + prev = saw_fast[-1] + saw_fast = [] + counter += 1 + + elif i["type"] == "FAST": + saw_fast.append(i["val"]) + else: + assert False + + # check the opposite, hard on fast, soft on slow + + print("SLOW SOFT check") + + fast_out = output[FAST3_SENSOR_DTYPE].output + slow_out = output[slow3_w].output + sorted_recvs = {} + + for fast_item, slow_item in zip(fast_out, slow_out): + # get fast window + print( + fast_item["data"].data["sensor"], + fast_item["data"].data["sensor_time"], + fast_item["recv_time"], + slow_item["data"].data, + slow_item["recv_time"], + ) + + # sort by recv time. + assert len(slow_item["data"].data) == 1 + val = slow_item["data"].data[0]["sensor"] + + sorted_recvs[slow_item["recv_time"]] = {"val": val, "type": "SLOW"} + + val = fast_item["data"].data["sensor"] + sorted_recvs[fast_item["recv_time"]] = { + "val": val, + "type": "FAST", + } + + sorted_recvs = dict(sorted(sorted_recvs.items(), key=lambda item: item[0])) + + assert len(sorted_recvs) == N_ITERS * 2 + + # simulate expected: + + prev_type = "SLOW" + prev_slow = 0 + counter = 0 + for i in sorted_recvs.values(): + if i["type"] == "FAST": + assert prev_type == "SLOW" + prev_type = "FAST" + continue + + if i["type"] == "SLOW": + assert prev_type == "FAST" + assert i["val"] >= prev_slow + prev_slow = i["val"] + prev_type = "SLOW" + continue + + assert False + + # check soft only + + print("SOFT ONLY check") + + fast_out = output[fast4_w].output + slow_out = output[slow4_w].output + sorted_recvs = {} + + for fast_item, slow_item in zip(fast_out, slow_out): + # get fast window + print( + fast_item["data"].data, + fast_item["recv_time"], + slow_item["data"].data, + slow_item["recv_time"], + ) + + # sort by recv time. + assert len(slow_item["data"].data) == 1 + sorted_recvs[slow_item["recv_time"]] = { + "val": slow_item["data"].data[0]["sensor"], + "type": "SLOW", + } + + assert len(fast_item["data"].data) == 1 + sorted_recvs[fast_item["recv_time"]] = { + "val": fast_item["data"].data[0]["sensor"], + "type": "FAST", + } + + sorted_recvs = dict(sorted(sorted_recvs.items(), key=lambda item: item[0])) + + # simulate expected: + + prev_type = "SLOW" + prev_slow = 0 + counter = 0 + for i in sorted_recvs.values(): + if i["type"] == "FAST": + assert prev_type == "SLOW" + prev_type = "FAST" + continue + + if i["type"] == "SLOW": + assert prev_type == "FAST" + assert i["val"] >= prev_slow + prev_slow = i["val"] + prev_type = "SLOW" + continue + + assert False + + # check data split + print("Check data split") + + assert len(pos_monitor.output) == len(neg_monitor.output) + # check vals + for p, n in zip(pos_monitor.output, neg_monitor.output): + assert p.data["sensor"] >= 0 and p.data["sensor"] % 2 == 0 + assert n.data["sensor"] < 0 and n.data["sensor"] % 2 == 1 + print(p.data["sensor"], n.data["sensor"]) + + # done!