DeviceHub

16 min

The DeviceHub module manages devices, tags, drivers, and asset discovery on Litmus Edge.

Imports

from litmussdk.devicehub import devices, tags, drivers

Devices

Listing Devices

all_devices = devices.list_devices()  # list[Device]

# Get a single device by ID
device = devices.list_device_by_id(device_id)

Pass raw=True to skip pydantic validation and get the API JSON unchanged. Useful when the device schema drifts ahead of your installed SDK or when you only need a few fields.

all_raw = devices.list_devices(raw=True)         # list[dict[str, Any]]
device_raw = devices.list_device_by_id(device_id, raw=True)

Each Device is a Pydantic model with fields including:

id: str | None

name: str

driver: Driver

settings: dict

alias_topics: bool

debug: bool

dh_params: dict

worker_params: dict

# Find a device by name
device = next((d for d in devices.list_devices() if d.name == "My Device"), None)
device_id = device.id
driver = device.driver

Creating a Device

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

from litmussdk.devicehub import devices
from litmussdk.devicehub.devices import Device
from litmussdk.devicehub.record import DriverTemplates  # bundled driver templates

simulator = DriverTemplates.simulator_gen1

template = Device(
    name="My Simulator",
    driver=simulator,                  # Driver object
    settings={},                       # varies by driver, see Driver Templates
    alias_topics=True,
    debug=False,
    dh_params={"WatchDog": "false"},
    worker_params={"PublishOnPollingInterval": "false"},
)
created = devices.create_device(template)
device_id = created.id

Calling create_device on a Device that already has an id raises ValueError.

Updating a Device

update_device also takes a Device object. Fetch the existing device, mutate it, push it back. The Device's id must be set.

existing = devices.list_device_by_id(device_id)
existing.name = "Renamed Simulator"
existing.settings = {}
updated = devices.update_device(existing)

Note: a device's driver cannot be changed after creation.

Deleting Devices

devices.delete_device(device)                        # takes a Device object

# Delete multiple at once by id
devices.delete_devices_by_ids([id1, id2])

# Delete multiple at once by Device objects
devices.delete_devices([device_obj_1, device_obj_2])

Starting and Stopping

devices.start_devices([device_obj_1, device_obj_2])  # list of Device objects
devices.stop_devices([device_obj_1, device_obj_2])

# Single device
devices.start_device(device)
devices.stop_device(device)

Calling start_devices on an already-running device (or stop_devices on a stopped one) will raise an error.

Tags

Listing Tags

all_tags = tags.list_all_tags()                    # list[Tag]

# list_registers_from_single_device takes a Device object, not an id
device = devices.list_device_by_id(device_id)
device_tags = tags.list_registers_from_single_device(device)  # list[Tag]

Both functions accept raw=True to skip pydantic validation and return the API JSON as-is.

all_raw = tags.list_all_tags(raw=True)                              # list[dict[str, Any]]
device_raw = tags.list_registers_from_single_device(device, raw=True)

Each Tag is a Pydantic model with fields including:

id: str | None (None until the tag is created)

device: Device (parent device, required)

name: str (register type, e.g. "S", "int", "M" - constrained by the device's driver)

tag_name: str (display name)

value_type: str

description: str

properties: KeyedList (configuration key-value pairs validated against the driver's register type)

topics: list[Topic] (auto-generated on creation)

publish_cov: bool

metadata: KeyedList

Creating Tags

create_tags takes a list of Tag objects. Each tag must have a device reference; do not set id (Litmus Edge generates it).

from litmussdk.devicehub.tags import Tag

device = devices.list_device_by_id(device_id)

tag = Tag(
    device=device,
    name="S",                          # register type, must match a SupportedRegister of the driver
    tag_name="Pressure",
    value_type="Float64",
    properties={                       # KeyedList of property name -> value
        "PollingInterval": "1000",
        "Address": "1",
    },
)

created = tags.create_tags([tag])      # list[Tag] with ids populated
new_id = created[0].id

Updating Tags

update_tags also takes a list of Tag objects. Each Tag must have its id set. Mutate fields and push back.

existing_tags = tags.list_registers_from_single_device(device)
target = next(t for t in existing_tags if t.tag_name == "Pressure")
target.tag_name = "Pressure Renamed"
updated = tags.update_tags([target])

All fields on the Tag are sent. Mutating a single field is sufficient; other fields keep their existing values.

Deleting Tags

delete_tags takes a list of Tag objects, not IDs.

device = devices.list_device_by_id(device_id)
device_tags = tags.list_registers_from_single_device(device)
tags.delete_tags(device_tags)          # delete all tags on the device