Example Scripts

Check Battery

"""
Sample script for using the Guardian Earbud Client

Checking battery level
"""

import asyncio

from idun_guardian_sdk import GuardianClient


if __name__ == "__main__":
    client = GuardianClient()

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

check_battery.py

Check Impedance

"""
Sample script for using the Guardian Earbud Client

Checking impedance
"""

import asyncio

from idun_guardian_sdk import GuardianClient


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


if __name__ == "__main__":
    client = GuardianClient(debug=True)
    asyncio.run(client.stream_impedance(handler=print_impedance))

check_impedance.py

Check Impedance For a Specific Time

"""
Sample script for using the Guardian Earbud Client

Checking impedance with a sample task running in parallel and stop_impedance command

Also demonstrates the connect and disconnect methods (not required to be called)
"""

import asyncio
from idun_guardian_sdk import GuardianClient


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


async def sample_task(seconds):
    """
    Replace your function/task here
    """
    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()  # Connect to the device is not required, sdk handles it

    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()  # Disconnect to the device is not required to be called, sdk handles it on program exit

    except (KeyboardInterrupt, asyncio.CancelledError):
        pass


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

check_impedance_start_stop.py

Delete Recording

"""
Sample script for using the Guardian Earbud Client

Delete a recording
"""

from idun_guardian_sdk import GuardianClient


my_api_token = ""
my_recording_id = ""


if __name__ == "__main__":

    client = GuardianClient(api_token=my_api_token)
    recordings = client.delete_recording(recording_id=my_recording_id)

delete_recording.py

Download Data

"""
Sample script for using the Guardian Earbud Client

Download EEG data from a recording
"""

from idun_guardian_sdk import GuardianClient, FileTypes


my_api_token = ""
my_recording_id = ""


if __name__ == "__main__":
    client = GuardianClient(api_token=my_api_token)
    client.download_file(recording_id=my_recording_id, file_type=FileTypes.EEG)

download_data.py

Download Daytime Report

"""
Sample script for using the Guardian Earbud Client

Generate and Download Daytime Report from a recording
"""

from idun_guardian_sdk import GuardianClient


my_api_token = ""
my_recording_id = ""


if __name__ == "__main__":
    client = GuardianClient(api_token=my_api_token)
    client.generate_and_download_daytime_report(recording_id=my_recording_id)

download_daytime_report.py

Download Sleep Report

"""
Sample script for using the Guardian Earbud Client

Generate and Download Sleep Report from a recording
"""

from idun_guardian_sdk import GuardianClient


my_api_token = ""
my_recording_id = ""


if __name__ == "__main__":
    client = GuardianClient(api_token=my_api_token)
    client.generate_and_download_sleep_report(recording_id=my_recording_id)

download_sleep_report.py

Get Recordings

"""
Sample script for using the Guardian Earbud Client

Get a list of recordings
"""

from datetime import datetime

from idun_guardian_sdk import GuardianClient


my_api_token = ""


def to_date(timestamp):
    """
    Converts timestamp to formatted date/time
    """
    milliseconds = int(timestamp / 1000)
    dt = datetime.fromtimestamp(milliseconds)
    formatted_date = dt.strftime("%Y-%m-%dT%H:%M:%S")
    return formatted_date


if __name__ == "__main__":

    client = GuardianClient(api_token=my_api_token)
    recordings = client.get_recordings(status="COMPLETED", limit=10)

    for recording in recordings["items"]:
        recording_date = to_date(recording["recordingId"])
        recordin_id = recording["recordingId"]
        print(f"Recording Date: {recording_date}. Recoring ID: {recordin_id}")

get_recordings.py

Record Data

Note

The realtime predictions feature is not enabled by default. If you want to test this, get in touch with us.

"""
Sample script for using the Guardian Earbud Client

- Start recording data from the Guardian Earbuds
"""

import asyncio
from idun_guardian_sdk import GuardianClient

RECORDING_TIMER: int = 60 * 60 * 2  # 2 hours of recording
LED_SLEEP: bool = False

my_api_token = ""


# Example callback function
def print_data(event):
    print("CB Func:", event.message)


if __name__ == "__main__":
    client = GuardianClient(api_token=my_api_token, debug=True)

    # Subscribe to live insights and/or realtime predictions
    client.subscribe_live_insights(raw_eeg=True, filtered_eeg=True, handler=print_data)
    client.subscribe_realtime_predictions(fft=True, jaw_clench=False, handler=print_data)

    # start a recording session
    asyncio.run(
        client.start_recording(
            recording_timer=RECORDING_TIMER, led_sleep=LED_SLEEP, calc_latency=False
        )
    )

record_data.py

Record Data and Subscribe to Realtime Insights and Predictions

Note

The realtime predictions feature is not enabled by default. If you want to test this, get in touch with us.

"""
Sample script for using the Guardian Earbud Client

Full example of recording data, generating reports and downloading the files
"""

import asyncio

from idun_guardian_sdk import GuardianClient, FileTypes


my_api_token = ""
RECORDING_TIMER = 60 * 15  # 15 min


def print_data(data):
    print(data.message)


if __name__ == "__main__":
    client = GuardianClient(api_token=my_api_token)

    client.subscribe_live_insights(raw_eeg=True, filtered_eeg=True, handler=print_data)
    client.subscribe_realtime_predictions(
        fft=True, jaw_clench=False, quality_score=True, handler=print_data
    )

    asyncio.run(client.start_recording(recording_timer=RECORDING_TIMER))
    rec_id = client.get_recording_id()

    print("RecordingId", rec_id)
    client.update_recording_tags(recording_id=rec_id, tags=["tag1", "tag2"])
    client.update_recording_display_name(recording_id=rec_id, display_name="todays_recordings")
    client.download_file(recording_id=rec_id, file_type=FileTypes.EEG)
    client.generate_and_download_sleep_report(recording_id=rec_id)
    client.generate_and_download_daytime_report(recording_id=rec_id)

record_data_full_example.py

Stream Data to LSL

Note

This requires pylsl to be installed.

import asyncio
from pylsl import StreamInfo, StreamOutlet

from idun_guardian_sdk import GuardianClient

RECORDING_TIMER: int = 60 * 60 * 2  # 2 hours
my_api_token = ""


if __name__ == "__main__":
    client = GuardianClient(api_token=my_api_token)
    client.address = asyncio.run(client.search_device())

    info = StreamInfo("IDUN", "EEG", 1, 250, "float32", client.address)
    lsl_outlet = StreamOutlet(info, 20, 360)

    def lsl_stream_handler(event):
        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.subscribe_live_insights(
        raw_eeg=True,
        handler=lsl_stream_handler,
    )

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

stream_to_lsl.py

Stream Data to LSL and Check Battery

Note

This requires pylsl to be installed.

import asyncio
from pylsl import StreamInfo, StreamOutlet

from idun_guardian_sdk import GuardianClient

RECORDING_TIMER: int = 60 * 60 * 2  # 2 hours
my_api_token = ""


async def stop_task(task):
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass


async def read_battery_forever(client):
    while True:
        battery = await client.check_battery()
        print("Battery Level: %s%%" % battery)
        await asyncio.sleep(1)


async def main():
    client = GuardianClient(api_token=my_api_token, debug=True)
    client.address = await client.search_device()

    info = StreamInfo("IDUN", "EEG", 1, 250, "float32", client.address)
    lsl_outlet = StreamOutlet(info, 20, 360)

    def lsl_stream_handler(event):
        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.subscribe_live_insights(
        raw_eeg=True,
        handler=lsl_stream_handler,
    )

    recording_task = asyncio.create_task(client.start_recording(recording_timer=RECORDING_TIMER))
    battery_task = asyncio.create_task(read_battery_forever(client))
    await recording_task
    await stop_task(battery_task)


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

stream_to_lsl_show_battery.py