Getting started

About the IDUN Guardian SDK

What can I do with the IDUN Guardian SDK?

The IDUN Guardian earbud can record data for a specified duration. The recording can be started and stopped programmatically, and the data can be accessed after the recording is completed. The recording duration is specified in seconds (see Start a recording).

During the recording, you can subscribe to real-time insights and predictions to get live data from the earbud. Real-time insights include raw EEG, band-passed EEG, and IMU data. Real-time predictions include jaw clench, horizontal eye movements (HEOG), Quality Score, and FFT z-scores (see Subscribe to real time insights and predictions).

In addition, you can stream data to LSL (Lab Streaming Layer) to integrate data from the earbud with other devices or software (see Stream data to LSL).

After the recording is completed, you can access the recorded data and generate reports (see Post Recording).

How do I record data with the IDUN Guardian SDK?

The main class is GuardianClient, which provides methods to connect to the earbud, record data, and access real-time insights and predictions.

It can be initialized with the following main parameters:

Device address:

The address of the earbud you want to connect to. It has the form of 6 groups of 2 characters separated by :, for example A1:B2:C3:D4:E5:F6.

If not provided, the SDK will search for the earbud automatically (see Search the earbud manually).

API Token:

Your IDUN API token. This is required for any Cloud API operation, like downloading recordings or generating reports. It is a long string composed of letters and numbers, and starting with idun_, for example idun_1abc23def456gh78ij9k0lm1nop2q3rs.

You can set this as an environment variable or pass it as an argument to the GuardianClient constructor (see Setup your API Token).

The rest of the parameters and functionalities can be found in the class documentation.

What is the general process for recording data?

Please visit our user manual for a detailed guide on how to use the IDUN Guardian earbud.

The general process for recording data with the IDUN Guardian SDK is as follows:

Pre-recording:
- Assemble the earbud.
- Clean ears.
- Turn the device on, put it on and let it stabilize for 10 minutes.
- Measure impedance. It should be below 300 kOhms (see Check impedance values).
- Check the battery level (see Connect to the device).
Recording:
- Prepare your recording setup.
- Connect to the earbud.
- If necessary, subscribe to real-time insights and predictions (see Subscribe to real time insights and predictions).
- Start the recording for a specified duration (see Start a recording).
Post-recording:
- Download the recording (see Download recording).
- Generate reports if needed (see Generate Sleep Report for a Recording).

About Asynchronous Operations with Asyncio

The IDUN Guardian SDK uses asyncio to manage asynchronous operations. Asynchronous operations are non-blocking and allow you to run multiple tasks concurrently. For more information on asyncio, see the official documentation.

Basic usage of asyncio

Any GuardianClient operations needs to be run using the asyncio.run() function.

For example, to connect to the earbud:

import asyncio
from idun_guardian_sdk import GuardianClient

if __name__ == "__main__":
    client = GuardianClient()
    asyncio.run(client.connect_device())

Or to start a recording with live insights:

import asyncio
from idun_guardian_sdk import GuardianClient

def print_data(event):
    print("CB Func:", event.message)

if __name__ == "__main__":
    client = GuardianClient(api_token="my_api_token")
    client.subscribe_live_insights(raw_eeg=True, filtered_eeg=True, handler=print_data)
    asyncio.run(client.start_recording(recording_timer=60))

Multiple asynchronous operations

To run multiple asynchronous operations successively, you need to wrap the operations in an async function and use await to wait for the completion of each operation. Indeed, Bleak requires the same asyncio run loop to be used for all operations. The program will not work correctly if you call asyncio.run() multiple times (see this page for more details).

For example, to connect to the earbud, start a recording, and then disconnect:

import asyncio
from idun_guardian_sdk import GuardianClient

async def main():
    client = GuardianClient()
    await client.connect_device()
    await client.start_recording(recording_timer=60)
    await client.disconnect_device()

if __name__ == "__main__":
    asyncio.run(main())

Pre Recording

Search the earbud manually

To search for the earbud, you need to run the following command in your Python shell or script:

import asyncio
from idun_guardian_sdk import GuardianClient

client = GuardianClient()
device_address = asyncio.run(client.search_device())

Follow the steps in the terminal to search for the earbud with the name IGEB.

If there are more than one IGEB device in the area, you will be asked to select the device you want to connect to. A list such as below will pop up in the terminal:

For Windows:

----- Available devices -----
Index | Name | Address
----------------------------
0     | IGEB | XX:XX:XX:XX:XX:XX
1     | IGEB | XX:XX:XX:XX:XX:XX
2     | IGEB | XX:XX:XX:XX:XX:XX
----------------------------

For Mac OS:

----- Available devices -----
Index | Name | UUID
----------------------------
0    | IGEB | XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX
1    | IGEB | XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX
2    | IGEB | XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX
----------------------------

Enter the index number of the device you want to connect to.

For the next executions, you can save the address and initialize the GuardianClient with address parameter:

import asyncio
from idun_guardian_sdk import GuardianClient

device_address = ""  # add address string here
client = GuardianClient(address=device_address)

If the address is not provided and you start connection to the device, the search_device() prompt will be automatically started.

Setup your API Token

Cloud API operations require an IDUN API key. The SDK calls this key an API token.

Get your API key

Open Guardian Console and log in.
In Account > Devices, link your Guardian device if it is not already linked.
Open Account > API key.
Select Generate API key if no key exists, then copy the key. If a key already exists, copy that key.

Note

Creating a new key requires a device linked to your account. Key creation checks the saved account link. The device does not need to be powered on or connected at this step.

Configure and verify your API key

Set the IDUN_API_TOKEN environment variable in the environment that runs your application:

export IDUN_API_TOKEN="YOUR_API_KEY"

Replace YOUR_API_KEY locally with the key you copied. GuardianClient reads this environment variable automatically. Verify authentication without connecting to the device:

from idun_guardian_sdk import GuardianClient

client = GuardianClient()
client.get_user_info()
print("Authentication successful")

You can also pass a key loaded from your environment or secret store through the api_token argument to GuardianClient.

Warning

Keep your API key secret. Do not paste it into an AI agent chat, print it in logs, or commit it to source control. If you store it in a local file, exclude that file from source control. The agent only needs to know where the key is configured.

Connect to the device

To connect to the earbud, run the following command in your Python shell or script:

import asyncio
from idun_guardian_sdk import GuardianClient

client = GuardianClient()
asyncio.run(client.connect_device())

Check battery level

To read out the battery level, run the following command in your Python shell or script:

import asyncio
from idun_guardian_sdk import GuardianClient

client = GuardianClient()

battery_level = asyncio.run(client.check_battery())
print("Battery Level: %s%%" % battery_level)

Check impedance values

To read out the impedance values, run the following commands in your Python shell or script. Note that stream_impedance will run until you manually stop it by pressing Ctrl + C:

import asyncio
from idun_guardian_sdk import GuardianClient

MAINS_FREQUENCY_60Hz = False
client = GuardianClient()

asyncio.run(client.stream_impedance(mains_freq_60hz=MAINS_FREQUENCY_60Hz))

mains_freq_60hz is a boolean parameter that sets the mains frequency to 60Hz (for the US).

If you need to start/stop streaming impedance programmatically, you can create a coroutine task:

import asyncio
from idun_guardian_sdk import GuardianClient

def print_impedance(data):
    print(f"{data}\tOhm")

async def sample_task(seconds):
    for i in range(seconds):
        print(f"Run a sample task for: {i}/{seconds} seconds")
        await asyncio.sleep(1)

async def main():
    client = GuardianClient(debug=True)
    await client.connect_device()
    print("Starting impedance task")
    task = asyncio.create_task(client.stream_impedance(handler=print_impedance))

    try:
        await sample_task(seconds=15)
        client.stop_impedance()
        await task
        await client.disconnect_device()
    except (KeyboardInterrupt, asyncio.CancelledError):
        pass

if __name__ == "__main__":
    asyncio.run(main())

Recording

Start a recording

For one application recording sequentially, use start_recording() directly. Wait for it to return before starting the next recording. No cleanup override is needed.

Note

Starting a recording can end existing ONGOING and NOT_STARTED recordings on the account. Do not start while another client is recording on the same account.

To start a recording with a pre-defined timer, you need to provide the recording duration in seconds. For example, for a 2 hours recording, you can use the following code:

import asyncio
from idun_guardian_sdk import GuardianClient

RECORDING_TIMER = 60 * 60 * 2  # 2 hours
LED_SLEEP = False

my_api_token = "idun_xxx"
my_device_address = "XX:XX:XX:XX:XX:XX"

def print_data(event):
    print("CB Func:", event.message)

client = GuardianClient(api_token=my_api_token, address=my_device_address)
asyncio.run(client.start_recording(recording_timer=RECORDING_TIMER, led_sleep=LED_SLEEP))

The default value for the recording timer is 10 hours (36000 seconds).

To stop the recording, wait for the timer or interrupt it with Ctrl+C.

Stream data to LSL

LSL (Lab Streaming Layer) is a system for synchronizing streaming data from different sources. It can be used to integrate data from the IDUN Guardian earbud with other devices or software.

To stream data to LSL, you need to install the pylsl library.

The process is similar to subscribing to real-time insights and predictions, but the handler function should be modified to stream the data to LSL.

Here is an example of how to stream raw EEG data to LSL:

import asyncio
from idun_guardian_sdk import GuardianClient
from pylsl import StreamInfo, StreamOutlet

device_address = "XX:XX:XX:XX:XX:XX"
my_api_token = "idun_xxx"

def stream_to_lsl(data):
    message = event.message
    eeg = message["raw_eeg"]
    most_recent_ts = eeg[-1]["timestamp"]
    data = [sample["ch1"] for sample in eeg]
    lsl_outlet.push_chunk(data, most_recent_ts)

client = GuardianClient(api_token=my_api_token, address=device_address)
client.subscribe_live_insights(raw_eeg=True, handler=stream_to_lsl)

# Create LSL stream
info = StreamInfo("IDUN", "EEG", 1, 250, "float32", device_address)
outlet = StreamOutlet(info, 20, 360)

asyncio.run(client.start_recording(recording_timer=60))

Post Recording

Get all recorded info

get_recordings() returns a dictionary. Read recordings from items and pass lastEvaluatedKey unchanged as page_index to fetch the next page. limit controls the page size, not the total number of results.

from idun_guardian_sdk import GuardianClient

client = GuardianClient()  # Reads IDUN_API_TOKEN from the environment.
page_index = None
while True:
    page = client.get_recordings(status="COMPLETED", limit=10, page_index=page_index)
    for recording in page["items"]:
        print(recording["recordingId"])
    page_index = page.get("lastEvaluatedKey")
    if not page_index:
        break

Recording IDs can be integers in list responses and strings when returned by start_recording(). Compare str(recording["recordingId"]) with str(recording_id) when locating a recording.

Download recording

Finishing await client.start_recording(...) ends capture. The cloud can still be processing the data. Wait for your recording to reach COMPLETED before requesting an export. Other statuses are NOT_STARTED, ONGOING, PROCESSING, and FAILED.

The example below checks all pages for a specific recording and stops on failure. It uses a two-minute polling deadline, checked after each full scan. Synchronous HTTP requests can exceed this deadline. A timeout does not mean the recording failed; check its status again later. It does not start a new recording.

import time
from idun_guardian_sdk import GuardianClient, FileTypes

client = GuardianClient()  # Reads IDUN_API_TOKEN from the environment.
recording_id = "YOUR_RECORDING_ID"  # Or the ID returned by start_recording().
deadline = time.monotonic() + 120
while True:
    status = None
    page_index = None
    while True:
        page = client.get_recordings(limit=200, page_index=page_index)
        for recording in page["items"]:
            if str(recording["recordingId"]) == str(recording_id):
                status = recording["status"]
                break
        page_index = page.get("lastEvaluatedKey")
        if status is not None or not page_index:
            break
    if status == "COMPLETED":
        break
    if status == "FAILED":
        raise RuntimeError("Recording processing failed")
    if time.monotonic() >= deadline:
        raise TimeoutError("Recording is not ready; check its status later")
    time.sleep(5)

client.download_file(
    recording_id=str(recording_id), file_type=FileTypes.EEG, file_path="downloads"
)

file_path is a destination directory, not a filename. The SDK creates it if needed and uses the filename supplied by the export URL. With no directory, it saves in the current working directory. download_file() returns None; do not use its return value as a path. An existing file with the same name is overwritten. EEG CSV exports contain timestamp and ch1 columns.

This example uses synchronous HTTP calls. In an async application, run this post-processing step outside the event loop or use asyncio.to_thread. Keep diagnostic output private: download exceptions can include signed URLs.

Generate Sleep Report for a Recording

To generate a sleep report:

from idun_guardian_sdk import GuardianClient

my_api_token = ""
my_recording_id = ""

client = GuardianClient(api_token=my_api_token)
client.generate_and_download_sleep_report(recording_id=my_recording_id)

Generate Daytime Report for a Recording

To generate a daytime report:

from idun_guardian_sdk import GuardianClient

my_api_token = ""
my_recording_id = ""

client = GuardianClient(api_token=my_api_token)
client.generate_and_download_daytime_report(recording_id=my_recording_id)

Disconnect from the device

To disconnect from the earbud, run the following command in your Python shell or script:

import asyncio
from idun_guardian_sdk import GuardianClient

client = GuardianClient()
asyncio.run(client.disconnect_device())