Litmus Edge Manager (LEM)

the lem module manages multi device deployments through the litmus edge manager companies, projects, devices, applications, alerts, certificates, ai models, and more imports from litmussdk lem import ( companies, dtwin, ) from litmussdk lem lifecycle import ( activation, aimodels, alerts, applications, certificates, dashboard, edgedevs, events, licenses, mpc, settings, templates, ) from litmussdk lem lifecycle edgedevs import general, details connection lem uses a separate connection object configured via the edge manager url and lem api token see managing connections for details export edge manager url=https //\<your lem ip> export edge api token=\<lem api token> all lem functions accept an optional connection lemconnection parameter if omitted the default connection (from environment variables) is used companies and projects all companies = companies list companies() # list\[company] stats = companies list all company stats() # list\[companystat] company = companies create company( company name="acme", description="acme corp devices", ) company = companies get company details("acme") companies delete company("acme") # also deletes all projects projects = companies get company projects("acme") # list\[project] project = companies get project details(project id) tz = companies get project timezone(project id) policy = companies get password policy() devices listing devices from litmussdk lem lifecycle edgedevs import general devices = general get all devices(project id) # list\[edgedevicecomplete] page = general get devices paginated(project id, page=0, limit=20) device = general get current device details(project id, device id) device id = device id hostname = device hostname device tags and versions tags = general get device tags(project id) versions = general get device versions(project id) license expiry expiring = general get license expiry in x days(project id, expiry days=30) expired = general get expired licenses(project id) applications application operations are task based they dispatch a job to the device and optionally wait for completion from litmussdk lem lifecycle import applications from litmussdk lem lifecycle applications import applicationparams params = applicationparams( app id="app uuid", name="my app", version="1 0 0", catalog id="catalog uuid", catalog name="my catalog", catalog url="https //catalog example com", ) success = applications launch application on device( name="deploy my app", company name="acme", project id=project id, device id=device id, application params=params, wait for online=true, timeout=120, ) apps = applications list device applications(project id, device id) app id = apps\[0] id applications start device app("start my app", "acme", project id, device id, app id) applications stop device app("stop my app", "acme", project id, device id, app id) applications delete device app("del my app", "acme", project id, device id, app id) async mode pass async =true to get back an asyncio task\[bool] instead of blocking import asyncio task = applications launch application on device( , async =true) success = await task activation and zerotouch from litmussdk lem lifecycle import activation from datetime import datetime, timezone config = activation add node locked activation config( project id=project id, expires at=datetime(2026, 12, 31, tzinfo=timezone utc), num devs=5, lease ttl days=365, ) activation code = config activation code configs = activation get activation config list(project id) config = activation get activation config(project id, activation code) requests = activation list all activation requests(project id) activation approve activation request(project id, entry id=requests\[0] entry id, accept=true) activation retry failed activation request(project id, entry id=requests\[0] entry id) activation remove config(activation code, project id) zerotouch zt = activation create zerotouch dynamicip( project id=project id, config name="zt pump floor", activation code=activation code, docker containers={"my app" "docker run d my app\ latest"}, ) zt = activation create zerotouch staticip( project id=project id, config name="zt pump floor static", activation code=activation code, docker containers={}, interface="eth0", ip="192 168 1 50", gateway="192 168 1 1", wan=false, ) activation remove zerotouch config(project id, zt id) alerts and triggers from litmussdk lem lifecycle import alerts from litmussdk lem lifecycle alerts import ( emailaction, trigger, eventrule, metricrule, ) actions = alerts get project actions(project id) email action = alerts create project action( project id=project id, action=emailaction( id=none, name="ops email", recipients=\["ops\@example com"], subject="alert {trigger name}", message="{message}", ), ) alerts delete project action(project id, email action id) triggers = alerts get project triggers(project id) trigger = alerts create trigger( project id=project id, name="high temp", subject="high temperature", message="temperature exceeded threshold", scope type="user", rules=\[ metricrule(template code="cpu usage", target value=90), ], active=true, severity=2, actions=\[email action id], ) alerts delete trigger(project id, trigger id) incidents = alerts list all incidents( project id=project id, state="open", ) certificates from litmussdk lem lifecycle import certificates certs = certificates get all certificates(project id) page = certificates get certificates paginated( project id=project id, order by="expiresafter asc", is online=true, limit=20, ) cert id = certificates issue new certificate( common name="pump 01", project id=project id, device id=device id, ttl days=365, authority="litmus", ) with open("cert pem", "rb") as cert f, open("ca pem", "rb") as ca f, open("key pem", "rb") as key f certificates upload custom certificate(project id, device id, cert f, ca f, key f) certificates reset certificate(project id, device id) ai / ml models from litmussdk lem lifecycle import aimodels with open("model tflite", "rb") as f aimodels upload new ml model(project id, file=f, description="pump anomaly detector") success = aimodels deploy ml model( name="deploy model", company name="acme", project id=project id, device id=device id, filename="model tflite", wait for online=true, timeout=120, ) page = aimodels get models(project id) deployed = aimodels list deployed models(project id, device id) aimodels delete model(project id, "model tflite") aimodels delete deployed models(\["model tflite"], project id, device id) backups from litmussdk lem lifecycle edgedevs import details backups = details list all backups(project id, device id) details make device backup("backup pump 01", "acme", project id, device id) details restore from backup("restore pump 01", "acme", project id, device id, backup id=backups\[0] id) url = details get presigned s3 backup url(project id, device id, backups\[0] id) details delete backup(project id, device id, backups\[0] id) details delete backups(project id, device id, \[b id for b in backups]) licenses from litmussdk lem lifecycle import licenses summary = licenses get license state summary() used = licenses list used licenses() skus = licenses list license skus() features = licenses list license features() with open("licenses csv", "rb") as f licenses upload license csv(f) licenses email license report("acme", \["admin\@example com"]) ok = licenses verify device limits("acme", num devices=10) dashboard and metrics from litmussdk lem lifecycle import dashboard deploy info = dashboard deployment info() usage = dashboard dashboard usage(project id) full usage = dashboard get project usage data(project id) system time = dashboard get system time() metrics = dashboard get device metrics(project id, device ids=\[device id]) project alerts = dashboard get project alerts(project id) digital twins (lem side) from litmussdk lem import dtwin \# model classes live in dtwin models, not the package root from litmussdk lem dtwin models import ( dtwinstaticattribute, dtwindynamicattribute, dtwintransformation, ) models = dtwin get dtwin models("acme") # list\[dtwinmodel] model = dtwin create dtwin model("acme", name="pumpmodel", description="centrifugal pump") model = dtwin get model details(model id) static attrs = dtwin get model static attrs(model id) dynamic attrs = dtwin get model dynamic attrs(model id) transforms = dtwin get model transformations(model id) hierarchy = dtwin get hierarchy(model root node id) dtwin update model( company name="acme", model id=model id, create static attrs=\[ dtwinstaticattribute(id="new uuid", key="location", value="ontario") ], update static attrs=\[], delete static attrs=\[], create dynamic attrs=\[ dtwindynamicattribute(name="pressure", data type="number", unit="kpa", topic="pump pressure") ], update dynamic attrs=\[], delete dynamic attrs=\[], create transformations=\[], update transformations=\[], delete transformations=\[], save hierarchy=none, ) marketplace catalog (mpc) from litmussdk lem lifecycle import mpc from litmussdk lem lifecycle mpc import dockerregistryconfig, releasecomponent, releasevariable catalogs = mpc list catalogs() catalog id = catalogs\[0] id \# registries are attached to a catalog, not standalone mpc add registry( catalog id=catalog id, config=dockerregistryconfig(url="", username="user", password="pass"), ) apps = mpc list catalog applications(catalog id) app = mpc create application(catalog id, name="my app") \# releases are referenced by their name (string), not by id release = mpc create application release( catalog id=catalog id, application id=app id, name="v1 0 0", type="docker", ) mpc add release component( catalog id=catalog id, application id=app id, release name=release name, component=releasecomponent( ), ) \# add release variables takes a single variable per call, not a list mpc add release variables( catalog id=catalog id, application id=app id, release name=release name, variable=releasevariable( ), ) mpc submit release(catalog id, app id, release name) mpc delete release(catalog id, app id, release name) mpc delete application(catalog id, app id) events from litmussdk lem lifecycle import events from datetime import datetime, timezone, timedelta now = datetime now(timezone utc) page = events get system events( company name="acme", date range=(now timedelta(hours=24), now), project id=project id, ) device details from litmussdk lem lifecycle edgedevs import details services = details list services(project id, device id) details start service("start svc", "acme", project id, device id, services\[0] id) details stop service("stop svc", "acme", project id, device id, services\[0] id) device licenses = details list device licenses(project id, device id) details add license online(project id, device id, "license key") details start metrics pull(project id, device id) details stop metrics pull(project id, device id) enabled = details is forwarding enabled(project id, device id) details enable disable forwarding(project id, device id, enabled=true) details reboot device("reboot pump 01", "acme", project id, device id) details create support bundle("bundle pump 01", "acme", project id, device id) raw responses many list and detail functions accept raw=true to skip pydantic parsing and return the api json unchanged coverage of raw=true varies by function; check each function's signature in your ide or via litmus sdk cli list lem lifecycle \<module> parsed = general get devices paginated(project id) # page\[edgedevicecomplete] raw = general get devices paginated(project id, raw=true) # dict\[str, any] license copyright (c) litmus automation inc