Litmus Edge Manager (LEM)

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

export EDGE_MANAGER_URL=https://<your-lem-ip>
export EDGE_API_TOKEN=<lem-api-token>

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()

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

tags = general.get_device_tags(project_id)
versions = general.get_device_versions(project_id)

expiring = general.get_license_expiry_in_x_days(project_id, expiry_days=30)
expired = general.get_expired_licenses(project_id)

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)

import asyncio
task = applications.launch_application_on_device(..., async_=True)
success = await task

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)

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)

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=["[email protected]"],
        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",
)

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)

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)

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])

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", ["[email protected]"])
ok = licenses.verify_device_limits("acme", num_devices=10)

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)

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,
)

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)

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,
)

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)

parsed = general.get_devices_paginated(project_id)               # Page[EdgeDeviceComplete]
raw = general.get_devices_paginated(project_id, raw=True)        # dict[str, Any]