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 license copyright (c) litmus automation inc