Analytics

17 min

The analytics module provides access to the Litmus Edge analytics service: processors, groups, AI models, and variables.

Imports

# REST API (3.x and 4.0.x)
from litmussdk import analytics

# GraphQL API (4.0.x only)
from litmussdk.analytics import _gql_instances, _gql_models, _variables

Note: GraphQL functions raise UnsupportedVersionError when called against a 3.x device.

REST API (3.x and 4.0.x)

Listing

from litmussdk import analytics

# List all processor groups
groups = analytics.list_all_groups()         # list[str]

# List all processors
processors = analytics.list_all_processors() # list[dict]

# List all available functions
functions = analytics.list_all_functions()   # list[dict]

Processor Library

The SDK ships with a local library of processor type definitions. Use it to look up processor configurations without an API call:

from litmussdk import analytics

# Get a processor type by name
internal_name, params = analytics.get_single_processor_from_library("Generator")

# Browse the full library
all_types = analytics.list_all_in_processor_library()

Creating a Processor

from litmussdk import analytics

internal_name, params = analytics.get_single_processor_from_library("Generator")

processor = analytics.create_processor(
    function_defined_name="my-generator",
    function_internal_name=internal_name,
    function_parameters=params,
    group_name="default",
)
processor_id = processor["ID"]

Reading and Updating a Processor

from litmussdk import analytics

processor_id = "969CA8AB-BF87-4537-B50E-57B109C40DBD"

# Get full processor details
proc = analytics.get_processor(processor_id)
params = proc["parameters"]

# Update a single parameter
params["strength"] = "100"
analytics.update_processor(processor_id, params)

# Get the last value published by the processor
value = analytics.get_processor_last_value(processor_id)

Enable / Disable

analytics.enable_processor(processor_id)   # start
analytics.enable_processor(processor_id, enabled=False)  # stop

Connecting Processors

connect_processor takes lists of processor IDs (one or many on each side) and a connection type.

analytics.connect_processor(
    connection_type="events",                # or "values"
    inputs_processors=[input_id],            # list, plural
    outputs_processors=[output_id],          # list, plural
)

Deleting

analytics.delete_processor(processor_id)

# Delete all processors in a group, use with caution
analytics.remove_all_processors()

GraphQL API (4.0.x only)

Groups

from litmussdk.analytics import _gql_instances

groups = _gql_instances.get_groups()                          # list[str]
instances = _gql_instances.get_processor_instances("default") # list[dict]

# Lock / unlock a group for editing
_gql_instances.set_group_lock_status("default", "CHECKOUT")
_gql_instances.set_group_lock_status("default", "CHECKIN")

Creating and Connecting Instances

# Create a processor instance in a group
_gql_instances.create_processor_instance("Generator", "my-gen")

# Connect instances
_gql_instances.connect_processor_instances(["src-id"], ["dst-id"], "values")

# Disconnect
_gql_instances.disconnect_processor_instances(["src-id"], ["dst-id"], "values")

Export / Import Groups

# Export a group configuration (use with AirGapTemplate)
payload = _gql_instances.export_group("default")

# Import a group configuration
_gql_instances.import_group("default", payload)

Variables and Parameters

from litmussdk.analytics import _variables

# Variables
_variables.get_analytics_variables()
_variables.set_analytics_variable("myVar", "42")

# Parameters
_variables.set_analytics_parameter("myParam", "val")

AI Models

from litmussdk.analytics import _gql_models

providers = _gql_models.get_ai_providers()  # list available providers
models = _gql_models.list_ai_models()

_gql_models.add_ai_model(
    provider="openai",
    name="gpt-4-model",
    url="https://api.openai.com",
    key="sk-...",
    model="gpt-4",
)
_gql_models.delete_ai_model("gpt-4-model")

# TensorFlow models
tf_models = _gql_models.get_tensorflow_models()
_gql_models.delete_tensorflow_model("my-tf-model")