dt-complete as a service: demo fixes, client feedback, deploy kit

use-cases/.gitignore

@@ -0,0 +1,7 @@
+__pycache__/
+tmp.py
+rhapsody.data.*
+workspace/
+__pycache__/
+dump.rdb
+.vscode/

use-cases/dt-complete/README.md

@@ -0,0 +1,47 @@
+
+# DT-Complete
+
+A demonstration of a complete digital twin:
+
+2 sensors + 3 surrogates* / physics entities
+
+Sensors:
+- M3DC1 Mock sensor
+- Random Value sensor
+
+Three physical entities:
+- M3DC1 Investigator
+- Runs a M3DC1 Investigator
+- Runs a DEMO_AGENT Agent (a simple pass through)
+- Runs a NEGATIVE_Agent Agent (simply computes the negative of sensor input)
+
+> *Technically, the M3DC1 trains two surrogates and then picks the best one.
+
+
+**Digital Twin Description Graph:**
+```
+
+M3DC1 Mock sensor --> M3DC1 Investigator --
+                                           \ 
+                                            --(JOIN)--> DEMO Agent --> OUT  
+RAND_VAL sensor  ---> NEGATIVE_Agent ------/
+
+```
+
+
+
+## To run:
+
+1. Install the digital twins library:
+
+``` bash
+git clone https://github.com/radical-cybertools/digital.twins
+
+# this is for the plain DT framework without all the as-a-service changes
+git checkout release/vanilla-framework
+pip install . 
+```
+
+2. Start up your sensors: `python3 m3dc1_mock_sensor.py` and `python3 rand_sensor.py`
+3. Start up the PUB/SUB streaming broker: `python3 local_broker.py`
+4. Finally, run `python3 run_me.py`

use-cases/dt-complete/demo_agent/demo_agent.py

@@ -0,0 +1,103 @@
+"""
+This agent is a demo of the "SciAgents" abstraction.
+
+This Demo_Agent stores all the models and their accuracies generated by both
+investigators. It then updates the model selector to always use the most
+accurate model.
+
+---
+More about SciAgents:
+
+A SciAgent is used to group together multiple investigators that operate on the
+same input / output DataTypes under one roof. It also has a "model selector"
+task that runs in-stream, deciding what investigator and model to run for
+inference.
+
+The purpose of the Science Agent is to contain one physics property. The
+investigator then provides the implementation.
+This implementation can have an Active Learner, and publishes one surrogate.
+
+
+The alternative is to have only an investigator, and put all the various
+surrogates inside one active learning loop. This is absolutely acceptable (see
+the m3dc1 investigator), though the SciAgent format is more generalizable and scalable.
+It separates the concerns from training a specific surrogate architecture from
+the decision making of what surrogate to train/run when.
+"""
+
+DO_PRINT = False
+import asyncio
+import json
+
+import pandas as pd
+from radical.asyncflow import WorkflowEngine
+from digitaltwin.components import ModelInvestigator, TypedData, SciAgent
+from digitaltwin.runtime import RuntimeAPI
+
+from .demo_investigator1 import Demo_Investigator_1
+from .demo_investigator2 import Demo_Investigator_2
+
+from dtypes import *
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class DEMO_Agent(SciAgent):
+    def __init__(self, flow: WorkflowEngine):
+        super().__init__(flow)
+        self.flow = flow
+
+        # no learning. Simple investigator
+        self.investigator_1 = Demo_Investigator_1(flow)
+        self.investigator_2 = Demo_Investigator_2(flow)
+
+        self.model_inventory: list[dict] = []
+        self.update_event = asyncio.Event()
+
+        @self.flow.function_task
+        async def model_select(in_data: TypedData, i_id=0, model_kwargs={}):
+            return i_id, model_kwargs
+
+        self.model_selector = model_select
+
+    async def model_publish_cb(
+        self, investigator: ModelInvestigator, model_args, acc_metrics
+    ):
+        # when a new model is published, add it to my model inventory
+        model = {
+            "investigator": investigator.get_id(),
+            "model_args": json.dumps(model_args),
+            "acc": acc_metrics.get("acc", 0),
+        }
+        self.model_inventory.append(model)
+        self.update_event.set()
+
+    async def main_loop(self, runtime: RuntimeAPI):
+        # Start up the investigator
+        runtime.start_investigator(self.investigator_1)
+        runtime.start_investigator(self.investigator_2)
+        runtime.subscribe_to_topic(runtime.ON_MODEL_PUBLISH, self.model_publish_cb)
+
+        runtime.set_model_selection_task(self.model_selector)
+
+        # default to investigator 1
+        runtime.update_model_selector(i_id=self.investigator_1.get_id())
+
+        while True:
+            await self.update_event.wait()
+            model_df = pd.DataFrame(self.model_inventory)
+
+            # select the model with the best accuracy.
+            sorted_df = model_df.sort_values(by="acc", ascending=False)
+            investigator_out = int(sorted_df.iloc[0]["investigator"])
+
+            m_args = json.loads(sorted_df.iloc[0]["model_args"])
+            if DO_PRINT:
+                print(
+                    f"[Demo Agent]: Winner is Investigator {investigator_out + 1}, model: {m_args}"
+                )
+            runtime.update_model_selector(i_id=investigator_out, model_kwargs=m_args)
+            del model_df
+            self.update_event.clear()

use-cases/dt-complete/demo_agent/demo_dtypes.py

@@ -0,0 +1,4 @@
+from digitaltwin.components import DataType
+
+# Demo Agent Prediction
+DEMO_PREDICTION = DataType("DEMO_PREDICTION")

use-cases/dt-complete/demo_agent/demo_investigator1.py

@@ -0,0 +1,113 @@
+"""
+The Demo Investigator is a simple investigator that triggers an active learning
+workflow per input on callback. It batches the input when the active learning
+workflow is running, so there is at most one workflow running at once.
+
+This is a simple example of how to interact with ROSE's AL inside the digital
+twin framework. This example shows how to have an active learner where the workflow is launched
+from the data stream.
+
+Though Demo Investigator 1 & 2 are identical, this is to demonstrate that I can
+have different "implementations" of a physics property. I can have each
+investigator focus on a single surrogate, have a single active learner loop, or
+other custom logic / lifecycle management. The DEMO_AGENT selects the
+investigator / surrogate to run.
+
+Note: Compare this to the M3DC1 investigator which instructs the simulation itself to wait for
+data. Therefore, the M3DC1 investigator requires a side-channel as the
+simulation is fetching the data. The approach here does not require REDIS or
+some side-channel method for sending data as the workflow is built already
+knowing the input data.
+
+"""
+
+DO_PRINT = False
+
+import asyncio
+import random
+from typing import Any
+import cloudpickle
+from digitaltwin import (
+    ModelInvestigator,
+    RuntimeAPI,
+    TypedData,
+)
+from radical.asyncflow import WorkflowEngine
+
+from rose.al.active_learner import Learner
+from .demo_dtypes import DEMO_PREDICTION
+
+
+class Demo_Investigator_1(ModelInvestigator):
+    def __init__(self, flow: WorkflowEngine):
+        super().__init__(flow)
+
+        # Learners
+        self.acl = Learner(flow)
+
+        self.data_update = asyncio.Event()
+        self.dataset: list[Any] = []
+        self.new_values: list[Any] = []
+
+        # Learning tasks..............
+        @self.acl.simulation_task(as_executable=False)
+        async def simulation(*args):
+            import time
+
+            time.sleep(1)
+            return time.time()
+
+        self.simulation = simulation
+
+        @self.acl.training_task(as_executable=False)
+        async def training(*args):
+            return random.random()
+
+        self.training = training
+
+        @self.flow.function_task
+        async def do_inference(in_data: TypedData, model=None):
+            # gamma = in_data.data[0].data
+            # neg = in_data.data[1].data
+
+            # out = [gamma, neg]
+            # if gamma is None:
+            #     out[0] = None
+
+            return TypedData(DEMO_PREDICTION, in_data.data)
+
+        self.inference = do_inference
+
+    async def input_callback(self, in_data: TypedData):
+        # only trigger update for ~10% of inputs
+        if random.random() > 0.1:
+            return
+        self.new_values.append(in_data)
+        self.data_update.set()
+
+    async def main_loop(self, runtime: RuntimeAPI):
+        # run the pipeline
+        runtime.subscribe_to_topic(runtime.ON_INPUT, self.input_callback)
+        runtime.set_inference_task(self.inference)
+        runtime.publish_new_model()
+        counter = 0
+        while True:
+            await self.data_update.wait()
+
+            self.dataset += self.new_values
+            self.new_values = []
+
+            # Start the active learning workflow on the dataset.
+            if DO_PRINT:
+                print("[Demo Agent / Investigator 1]: Start AL Workflow")
+            model = await self.training(self.simulation(self.dataset))
+
+            # publish model and accuracy metrics.
+            acc = random.random()
+            if DO_PRINT:
+                print(
+                    f"[Demo Agent / Investigator 1]: Publish model {counter}. Acc: {acc}"
+                )
+            runtime.publish_new_model({"model": counter}, {"acc": acc})
+            self.data_update.clear()
+            counter += 1

use-cases/dt-complete/demo_agent/demo_investigator2.py

@@ -0,0 +1,114 @@
+"""
+The Demo Investigator is a simple investigator that triggers an active learning
+workflow per input on callback. It batches the input when the active learning
+workflow is running, so there is at most one workflow running at once.
+
+This is a simple example of how to interact with ROSE's AL inside the digital
+twin framework. This example shows how to have an active learner where the workflow is launched
+from the data stream.
+
+Though Demo Investigator 1 & 2 are identical, this is to demonstrate that I can
+have different "implementations" of a physics property. I can have each
+investigator focus on a single surrogate, have a single active learner loop, or
+other custom logic / lifecycle management. The DEMO_AGENT selects the
+investigator / surrogate to run.
+
+Note: Compare this to the M3DC1 investigator which instructs the simulation itself to wait for
+data. Therefore, the M3DC1 investigator requires a side-channel as the
+simulation is fetching the data. The approach here does not require REDIS or
+some side-channel method for sending data as the workflow is built already
+knowing the input data.
+
+"""
+
+DO_PRINT = False
+
+import asyncio
+import random
+from typing import Any
+import cloudpickle
+from digitaltwin import (
+    ModelInvestigator,
+    RuntimeAPI,
+    TypedData,
+)
+from radical.asyncflow import WorkflowEngine
+
+from rose.al.active_learner import Learner
+from .demo_dtypes import DEMO_PREDICTION
+
+
+class Demo_Investigator_2(ModelInvestigator):
+    def __init__(self, flow: WorkflowEngine):
+        super().__init__(flow)
+
+        # Learners
+        self.acl = Learner(flow)
+
+        self.data_update = asyncio.Event()
+        self.dataset: list[Any] = []
+        self.new_values: list[Any] = []
+
+        # Learning tasks..............
+        @self.acl.simulation_task(as_executable=False)
+        async def simulation(*args):
+            import time
+
+            time.sleep(1)
+            return time.time()
+
+        self.simulation = simulation
+
+        @self.acl.training_task(as_executable=False)
+        async def training(*args):
+            return random.random()
+
+        self.training = training
+
+        @self.flow.function_task
+        async def do_inference(in_data: TypedData, model=None):
+            # gamma = in_data.data[0].data
+            # neg = in_data.data[1].data
+
+            # out = [gamma, neg]
+            # if gamma is None:
+            #     out[0] = None
+
+            return TypedData(DEMO_PREDICTION, in_data.data)
+
+        self.inference = do_inference
+
+    async def input_callback(self, in_data: TypedData):
+        self.new_values.append(in_data)
+
+        # only trigger update for ~10% of inputs
+        if random.random() > 0.1:
+            return
+        self.data_update.set()
+
+    async def main_loop(self, runtime: RuntimeAPI):
+        # run the pipeline
+        runtime.subscribe_to_topic(runtime.ON_INPUT, self.input_callback)
+        runtime.set_inference_task(self.inference)
+        runtime.publish_new_model()
+        counter = 0
+        while True:
+            await self.data_update.wait()
+
+            self.dataset += self.new_values
+            self.new_values = []
+
+            if DO_PRINT:
+                print("[Demo Agent / Investigator 2]: Start AL Workflow")
+            # Start the active learning workflow on the dataset.
+            model = await self.training(self.simulation(self.dataset))
+
+            # publish model and accuracy metrics.
+            acc = random.random()
+            if DO_PRINT:
+                print(
+                    f"[Demo Agent / Investigator 2]: Publish model {counter}. Acc: {acc}"
+                )
+            runtime.publish_new_model({"model": counter}, {"acc": acc})
+            self.data_update.clear()
+            counter += 1