Example: DeviceHub

15 min

End-to-end walkthrough: create a simulator device, add tags for pressure and flow rate, start it, read live tag values, then tear everything down.

Setup

from litmussdk.devicehub import devices, tags
from litmussdk.devicehub.devices import Device
from litmussdk.devicehub.tags import Tag
from litmussdk.devicehub.record import DriverTemplates

Assumes the default connection is configured; see Managing Connections.

1. Pick a Driver

Use the bundled driver template library so you don't need a live API call to look up driver IDs.

simulator = DriverTemplates.simulator_gen1
print(f"Driver: {simulator.name}, id={simulator.id}")

To browse the live driver list instead:

all_drivers = devices.drivers.list_all_drivers()  # list[Driver]
sim = next(d for d in all_drivers if d.name == "Simulator (Gen1)")

2. Create the Device

create_device takes a Device Pydantic object. Construct it without an id (Litmus Edge generates the id and returns the created Device).

template = Device(
    name="MyPump-01",
    driver=simulator,
    settings={},                                   # simulator gen1 needs no extra settings
    alias_topics=True,
    debug=False,
    dh_params={"WatchDog": "false"},
    worker_params={"PublishOnPollingInterval": "false"},
)

created = devices.create_device(template)
device_id = created.id
print(f"Created device {device_id}")

3. Start the Device

A newly created device is stopped. Start it before tags can publish values.

devices.start_device(created)

4. Add Tags

create_tags takes a list of Tag objects. Each tag must reference its parent Device. Don't set id (Litmus Edge generates it).

pressure = Tag(
    device=created,
    name="float",                                  # register type, must match a SupportedRegister
    tag_name="Pressure",
    value_type="Float64",
    properties={
        "PollingInterval": "1000",                 # ms
        "Address": "1",
    },
)

flow_rate = Tag(
    device=created,
    name="float",
    tag_name="FlowRate",
    value_type="Float64",
    properties={
        "PollingInterval": "1000",
        "Address": "2",
    },
)

new_tags = tags.create_tags([pressure, flow_rate])
for t in new_tags:
    print(f"Created tag {t.tag_name} ({t.id})")

device = devices.list_device_by_id(device_id)         # fetch fresh state
device_tags = tags.list_registers_from_single_device(device)
for t in device_tags:
    print(f"{t.tag_name}: type={t.name}, topics={[topic.topic for topic in t.topics]}")

Note: list_registers_from_single_device takes a Device object, not a string id.

6. Read Live Values

Tag values arrive via topics. The simulator publishes synthetic values at the polling interval.

The SDK doesn't subscribe to live topics directly; use the Integrations module to forward tag topics to a cloud connector, or query the underlying NATS / Litmus message bus from your own code. For a quick "did values arrive?" check, look at the analytics processor library which can consume tag topics directly:

from litmussdk import analytics

# Or check the raw API:
all_tags_raw = tags.list_all_tags(raw=True)            # list[dict[str, Any]]
print(f"Tag count across the device: {len(all_tags_raw)}")

7. Update a Tag

Mutate the existing Tag object and call update_tags. All fields are sent.

existing = tags.list_registers_from_single_device(device)
target = next(t for t in existing if t.tag_name == "Pressure")
target.tag_name = "Pressure (kPa)"
target.description = "Pump intake pressure"
updated = tags.update_tags([target])

8. Update the Device

device = devices.list_device_by_id(device_id)
device.name = "MyPump-01-renamed"
devices.update_device(device)

A device's driver cannot be changed after creation. Mutating driver is not allowed.

9. Tear Down

# Delete tags first
device = devices.list_device_by_id(device_id)
device_tags = tags.list_registers_from_single_device(device)
tags.delete_tags(device_tags)                          # takes list of Tag objects, not ids

# Stop and delete the device
devices.stop_device(device)
devices.delete_device(device)

Bulk Operations

For larger workloads:

# Get many devices
all_devices = devices.list_devices()
running = [d for d in all_devices if d.name.startswith("Pump-")]

# Stop them all in one call
devices.stop_devices(running)

# Delete them by id, faster than fetching each
devices.delete_devices_by_ids([d.id for d in running])

Importing Tags from CSV

For larger tag sets, edit a CSV and upload:

csv_text = tags.download_csv(device_id=device_id)      # current tags as CSV
# ... edit / generate CSV ...
tags.upload_csv(device_id=device_id, file_path="/path/to/new_tags.csv")