---
title: Example: DeviceHub
slug: sdk/example-usages/devicehub
docTags: 
createdAt: 2025-03-20T19:53:11.648Z
---

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

```python
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](docId\:vwtFo9SXcMRbFX9cAfENw).

## 1. Pick a Driver

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

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

To browse the live driver list instead:

```python
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).

```python
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.

```python
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).

```python
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})")
```

## 5. List Tags on the Device

```python
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]}")
```

:::BlockQuote
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](docId:-RvwUF5WrNpI9OKK3dhfb) 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:

```python
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.

```python
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

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

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

## 9. Tear Down

```python
# 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:

```python
# 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:

```python
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")
```

## See Also

- [DeviceHub](docId\:yxCOrVuyOocKWRIWiBFVd) - full reference for Device and Tag operations
- [DeviceHub Drivers](docId\:EhmsNvgYqkxU0IWKfGtQ4) - driver lookup, JSON schema validation
- [Example: Integration](docId\:RU9vOsgMpyWleRrmcmvr-) - send tag values to a cloud connector
- [Example: Digital Twins](docId\:wqjkxy4TkxtPkT6D0fGGV) - tie tags to a digital twin model