Example: Digital Twins

9 min

End-to-end walkthrough: build a Pump digital twin model, attach static and dynamic attributes, create an instance, and query live values.

Setup

from litmussdk import digital_twins
from litmussdk.devicehub import devices, tags

Assumes the default connection is configured; see Managing Connections.

1. Create a Model

model = digital_twins.create_model(
    name="Pump",
    description="Centrifugal pump - generic template",
)
model_id = model.id
print(f"Created model {model_id}")

2. Attach Static Attributes

Static attributes don't change at runtime - manufacturer info, location, etc.

digital_twins.create_static_attributes(
    model_id=model_id,
    attributes=[
        {"key": "Manufacturer", "value": "ACME Pumps"},
        {"key": "MaxFlowRate", "value": "200 L/min"},
    ],
)

3. Attach Dynamic Attributes

Dynamic attributes pull live values from tags or topics. First find a relevant tag from your device:

all_devices = devices.list_devices()
pump_device = next(d for d in all_devices if d.name == "PlantFloor-Pump-01")

device_tags = tags.list_registers_from_single_device(pump_device)
pressure_tag = next(t for t in device_tags if t.tag_name == "Pressure")

Then create a dynamic attribute that subscribes to that tag's topic:

digital_twins.create_dynamic_attribute(
    model_id=model_id,
    name="Pressure",
    data_type="Number",
    unit="kPa",
    topic=pressure_tag.topics[0].topic,
)

4. Create an Instance

The model is a template; instances are concrete twins of real devices.

instance = digital_twins.create_instance(
    model_id=model_id,
    name="PlantFloor-Pump-01",
)

# Set running so the instance starts subscribing to its dynamic attribute topics
digital_twins.set_instance_state(instance.id, state="RUNNING")

5. Query Live Values

# All instances of this model
all_pumps = digital_twins.get_instance_by_model(model_id)
for inst in all_pumps:
    print(f"{inst.name} - state={inst.state}")

6. Add a Transformation (Optional)

digital_twins.create_transformations(
    model_id=model_id,
    transformations=[{
        "name": "PressureCategory",
        "formula": 'IF(Pressure > 150, "HIGH", "OK")',
    }],
)

7. Tear Down

digital_twins.delete_instance(instance.id)
digital_twins.delete_model(model_id)