Export Sensor and Group Assignments to a CSV

The User Experience Insight (UXI) Onboarding API allows you to simplify most of the tasks associated with assigning your sensors and agents to the proper groups and configuring them to test your networks and services. To use the API, your dashboard must be on the Greenlake Cloud Platform. Additionally, you must be using the group-based assignment feature.

You can find documentation for the Onboarding API here:

Download the OpenApi specification.

Prerequisites
You will need to create a personal API client for the User Experience Insight service manager on the Greenlake Cloud Platform. See the detailed instructions on creating your personal API client in the Greenlake developer guide.

Example Scenario
In this example, suppose we want to get a list of all our sensors, determine what groups they are in, and write the data to a CSV file. This can be useful for inventory management, reporting, or further analysis.

Endpoints
Here are the endpoints you will need:

/networking-uxi/v1alpha1/sensors

/networking-uxi/v1alpha1/groups

/networking-uxi/v1alpha1/sensor-group-assignments

Note: The API uses cursor-based pagination.

Putting it together
Here is a Python 3 example of this in action. This example uses Python requests to fetch the data and write it to a CSV file.

This example does the following:

Generate a temporary access token using the credentials of the personal API client.

Retrieve a list of all sensors.

Retrieve a list of all groups.

Retrieve a list of all sensor group assignments to see which sensors are assigned to which groups.

Finally, the sensor and group information is written to a CSV file.

WARNING: This example is for demonstration purposes only. This example lacks robust error handling and optimizations necessary for a production environment. Do not use the example code in a production environment.

import csv
from typing import Any

import requests
from oauthlib.oauth2 import BackendApplicationClient
from requests.auth import HTTPBasicAuth
from requests_oauthlib import OAuth2Session

# Global constants
AUTH_TOKEN_URL = "https://sso.common.cloud.hpe.com/as/token.oauth2"
UXI_API_BASE_URL = "https://api.capenetworks.com/networking-uxi/v1alpha1"

# Customer specific constants
CLIENT_ID = "<CLIENT_ID>"
CLIENT_SECRET = "<CLIENT_SECRET>"

def main() -> None:
    api_token = get_api_token()

    sensors = get_sensors(api_token=api_token)
    groups = get_groups(api_token=api_token)
    assignments = get_sensor_group_assignments(api_token=api_token)

    headers, csv_data = generate_csv_data(sensors=sensors, groups=groups, assignments=assignments)

    write_to_file(headers, csv_data)


def get_api_token() -> str:
    client = BackendApplicationClient(CLIENT_ID)
    oauth = OAuth2Session(client=client)
    auth = HTTPBasicAuth(CLIENT_ID, CLIENT_SECRET)
    token = oauth.fetch_token(token_url=AUTH_TOKEN_URL, auth=auth)

    return token["access_token"]


def get_sensors(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="sensors", api_token=api_token)


def get_groups(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="groups", api_token=api_token)


def get_sensor_group_assignments(api_token: str) -> list[dict[str, Any]]:
    return _get_items(path="sensor-group-assignments", api_token=api_token)

def _get_items(path: str, api_token: str) -> list[dict[str, Any]]:
    HEADERS = {
        "Authorization": f"Bearer {api_token}",
    }
    items: list[dict[str, Any]] = []
    next_cursor = None

    # This while loop exists to keep making requests until all data for the specified resource is fetched.
    while True:
        try:
            params = {"next": next_cursor} if next_cursor else {}
            response = requests.get(
                f"{UXI_API_BASE_URL}/{path}", headers=HEADERS, params=params
            )
            response.raise_for_status()
            response_body: dict[str, Any] = response.json()
            response_items: list[dict[str, Any]] = response_body.get("items", [])
            next_cursor = response_body.get("next")

            items.extend(response_items)

            if not next_cursor or not response_items:
                break
        except requests.exceptions.HTTPError as e:
            print(f"Error fetching items ({path}): {e}")
            break

    return items


def generate_csv_data(
    sensors: list[dict[str, Any]],
    groups: list[dict[str, Any]],
    assignments: list[dict[str, Any]],
) -> tuple[list[str], list[list[str]]]:
    groups_map = {group["id"]: group for group in groups}
    assignments_map = {
        assignment["sensor"]["id"]: assignment["group"]["id"] for assignment in assignments
    }

    headers = ["Sensor Id", "Sensor Name", "Sensor Serial", "Group Id", "Group Name"]
    csv_data: list[list[str]] = []

    for sensor in sensors:
        sensor_id: str = sensor["id"]
        sensor_name: str = sensor["name"]
        sensor_serial: str = sensor["serial"]
        row = [sensor_id, sensor_name, sensor_serial]

        group_id = assignments_map.get(sensor_id, None)
        if group_id:
            group = groups_map[group_id]
            group_name = group["name"]
            row.extend([group_id, group_name])
        else:
            row.extend(["NA", "NA"])

        csv_data.append(row)

    return headers, csv_data


def write_to_file(headers: list[str], csv_data: list[list[str]]) -> None:
    with open("sensor_group_details.csv", "w", newline="") as file:
        writer = csv.writer(file)
        writer.writerow(headers)

        for row in csv_data:
            writer.writerow(row)

if __name__ == "__main__":
    main()