Experimental Digital Twin Framework

Main set of features implemented:

• Model Investigator
• Utility Tasks
• Persistent Tasks
• Callbacks
• Simple ZMQ pubsub backend
• Graph builder
• Convert to a Python Package
• Several Tests / Examples
• Science Agents
• Request inference API on runtime
• Barrier
• Split
• Join
• Shared SIM / subtasks running on agent, accessible by all investigators
• Barrier working on remote
• Split working on remote
• Join working on remote

Running the unit tests:

1. pip install .[test,service]
2. pytest (or tox for all supported interpreters)

The unit tests start their own stream broker on a random port; no setup. The integration tests under test/integration bring up a real ORBIT broker and rhapsody endpoint and skip themselves when they cannot.

Running the demos:

1. pip install .
2. cd test/
3. In one terminal, run local_broker.py -- it prints the addresses it bound
4. In a second terminal, cd into the demo and run run_me.py
5. In a third terminal, in the same demo, run its sensor.py if it has one (01-start-inference-stop and 04-start-agent-stop do)

When running a demo: be sure to start the ZMQ PubSub broker!

The third terminal is the point, not an inconvenience: a sensor is an external entity. It is a process of its own with a lifetime of its own, it publishes JSON on a shared channel, and it knows nothing about twins. The twin binds that channel with runtime.add_input(dtype, channel), and a second twin binding the same channel receives the same messages -- which is how one instrument feeds many twins. Start and stop the sensor independently of the twin; neither cares.

Demos without a sensor.py produce their input inside the twin, which is what persistent components are still for: 06-agent-pi drives itself off a timer, and 07-barrier off several.

Every side resolves the broker addresses the same way: DT_STREAM_PUB_ADDR and DT_STREAM_SUB_ADDR, defaulting to tcp://127.0.0.1:5000 and :5001 (see digitaltwin.config). Set them in every terminal to move the broker.

Binding policy: the broker binds to loopback by default, and it must stay that way unless you know what you are doing -- twin-internal payloads are cloudpickled, so anyone who can reach the broker ports can execute code in every subscriber. A non-loopback bind needs an explicit configuration and a private/firewalled network. External channels are decoded with the codec their binding names: json (the default) and raw are safe to accept from a producer you do not control, cloudpickle is not.

Running it as a service (the dt ORBIT plugin)

digitaltwin.service exposes the framework as a long-running ORBIT plugin: one session per client, many independent twins per session, and twins that keep running while their client is away. Installing the package registers the plugin through the radical.orbit.plugins entry point, so a broker or endpoint only has to be told to host it.

pip install .[service]

# 1 - the broker, hosting the dt plugin
radical-orbit-broker.py --plugins default,dt

# 2 - a rhapsody endpoint: where the twins' tasks execute.  The notify
#     window costs 250 ms on every sequential prediction at its default
radical-orbit-endpoint.py -n dt_task_ep
#     ... started with:
#     RADICAL_ORBIT_RHAPSODY_NOTIFY_WINDOW=0
#     RADICAL_ORBIT_RHAPSODY_BACKEND=concurrent

The client:

from radical.orbit import EndpointRuntime
from digitaltwin.components import NULL_DTYPE, TRUTHY
from digitaltwin.service import register_user_modules

import my_components                       # not installed on the service
register_user_modules([my_components])

rt = EndpointRuntime()
rt.start(wait=True)

# 'broker' is the participant hosting dt; engine wiring is explicit
dt = rt.get_plugin('broker', 'dt', config={
    'engines': {'task': {'endpoint_name': 'dt_task_ep',
                         'backends': ['concurrent']}}})

twin = dt.create_twin()                    # polls until the twin is ready
dt.add_task(twin, dt.package(MySensor), TRUTHY, SENSOR, is_persistent=True)
dt.add_investigator(twin, dt.package(MyModel), SENSOR, PREDICTION)
dt.start(twin)

print(dt.twin_list())                      # the observation mechanism
answer = dt.get_inference(twin, TypedData(SENSOR, 5), PREDICTION)

dt.twin_close(twin)

The session outlives the client: reattach with rt.get_plugin('broker', 'dt', sid=<sid>) and the twins are still there. dt.admin_sessions() lists every session, twin, state and last error on the service -- which is how orphans are found and torn down. test/09-service/ is a complete worked example.

Three contract notes:

• The client and the service must run the same digitaltwin version (and compatible Python / cloudpickle): shipped component classes pickle the framework by reference. Every call carries those versions and the service rejects skew with a clear error rather than failing somewhere inside an unpickle.
• A task's arguments are cloudpickled, but its return value must be JSON-safe or bytes -- ORBIT's rhapsody plugin JSON-encodes results and stringifies anything else. Return plain values from @flow.function_task bodies and wrap them in TypedData in the component.
• Persistent components run inline on the service's event loop. Their bodies must be thin async glue publishing through runtime.stream, never @flow.function_tasks (the service warns when it sees one).

Binding policy for the service (R7)

The plugin runs its own DT stream broker, embedded, one per plugin and shared by every twin. It binds to loopback on a random port by default, and that default is the safe one: its payloads are cloudpickled, so anyone who can reach the XSUB/XPUB ports gets code execution in every subscriber -- weaker than the token-authenticated ORBIT channel around it.

A non-loopback bind is possible (DT_STREAM_PUB_ADDR / DT_STREAM_SUB_ADDR on the service host) but requires a deliberate decision and a firewalled or private network. Until the data plane moves inside ORBIT's authenticated channel, do not expose those ports -- including in demos.