Example: DeviceHub

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 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 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})") 5\ list tags on the device 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 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 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") 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