Collect data from a device

edge-impulse-daemon --api-key <your project API key>

import requests
import getpass
import json

URL_STUDIO = "https://studio.edgeimpulse.com/v1/api/"
PROJECT_ID = int(input('Enter your Project ID: '))
AUTH_KEY = getpass.getpass('Enter your API key: ')


def check_response(response, debug=False):
    if not response.ok:
        raise RuntimeError("⛔️ Error\n%s" % response.text)
    else:
        if debug:
            print(response)
        return response


def do_get(url, auth, debug=False):
    if debug:
        print(url)
    response = requests.get(url,
                            headers={
                                "Accept": "application/json",
                                "x-api-key": auth
                            })
    return check_response(response, debug)


def parse_response(response, key=""):
    parsed = json.loads(response.text)
    if not parsed["success"]:
        raise RuntimeError(parsed["error"])
    if key == "":
        return json.loads(response.text)
    return json.loads(response.text)[key]


def get_project(project_id, project_auth, debug=False):
    response = do_get(URL_STUDIO + str(project_id), project_auth)
    return parse_response(response, "project")


print("Project %s is accessible" % get_project(PROJECT_ID, AUTH_KEY)["name"])

# https://studio.edgeimpulse.com/v1/api/{projectId}/devices

def get_devices(project_id, project_auth, debug=False):
    response = do_get(URL_STUDIO + str(project_id) + "/devices", project_auth)
    return parse_response(response, "devices")


device_id = ""
for device in get_devices(PROJECT_ID, AUTH_KEY):
    # if device["remote_mgmt_connected"] and device["supportsSnapshotStreaming"]:
    if device["remote_mgmt_connected"]:
        device_id = device["deviceId"]
        print("Found %s (type %s, id: %s)" %
              (device["name"], device["deviceType"], device_id))
        break
if device_id == "":
    print(
        "Could not find a connected device that supports snapshot streaming!")

# https://studio.edgeimpulse.com/v1/api/{projectId}/device/{deviceId}/start-sampling

SAMPLE_CATEGORY = "testing"
SAMPLE_LENGTH_MS = 20000
SAMPLE_LABEL = "squat"

def do_post(url, payload, auth, debug=False):
    if debug:
        print(url)
    response = requests.post(url,
                             headers={
                                 "Accept": "application/json",
                                 "x-api-key": auth
                             },
                             json=payload)
    return check_response(response, debug)


def collect_sample(project_id, device_id, project_auth, debug=False):
    payload = {
        "category": SAMPLE_CATEGORY,
        # "Microphone", "Inertial", "Environmental" or "Inertial + Environmental"
        "sensor": "Inertial",
        # The inverse of frequency in Hz
        "intervalMs": 10,
        "label": SAMPLE_LABEL,
        "lengthMs": SAMPLE_LENGTH_MS
    }
    response = do_post(
        URL_STUDIO + str(project_id) + "/device/" + str(device_id) +
        "/start-sampling", payload, project_auth, debug)
    return parse_response(response, "id")


print("Sample request returned", collect_sample(PROJECT_ID, device_id, AUTH_KEY))