[Muse 2] BrainFlow ML로 집중도 구하기!!

이번껀 전문가가 아니기에 틀린 정보가 많을 수 있습니다.

이번에도 집중도 구하기다!

다만 이번엔 인공지능을 곁들인..

 

내가 이전 부터 인공지능이라고 말하기는 했지만.. 이번껀 머신러닝에 가깝다고 볼 수 있다.

보통 인공지능 하면 딥러닝을 생각하겠지만.. 여기서는 머신러닝이다.

뭐 인공지능 안에 머신러닝이 포함되어있으니 상관은 없겠지만..

 

이번에 적을 내용은 이전 글 보다는? 적을테고 쉬울 것이다.

대부분이 이미 만들어진 것을 쓰기 때문에..

하지만 자세하게 들어가면 역시 어렵긴 하지만.. 최대한 내가 이해한 배경으로 설명해보겠다..!

틀린 정보가 있을 수도 있다.. 작년인 대학교 1학년 때 했던 작업들이고.. 2~4개월 밖에 안했었기 때문에..

 

그리고.. 28일이 아닌 25 ~ 26일에 기기를 반납해야해서 서둘러야 한다..!!!

 

암튼 먼저 코드부터!

 

Brainflow ML로 집중도 구하는 코드 공유!

from time import sleep
from brainflow.board_shim import BoardShim, BrainFlowInputParams, BoardIds
from brainflow.data_filter import DataFilter
from brainflow.ml_model import MLModel, BrainFlowModelParams, BrainFlowMetrics, BrainFlowClassifiers

board_id = BoardIds.MUSE_2_BOARD
params = BrainFlowInputParams()
params.serial_number = "Muse-0465" # Muse 2의 고유 시리얼 넘버
board_shim = BoardShim(board_id, params)

use_data_seconds = 5 # n초 동안 데이터 수집 
sampling_rate = BoardShim.get_sampling_rate(board_id)
eeg_channels = BoardShim.get_eeg_channels(board_id)
num_points = use_data_seconds * sampling_rate # 몇 초 동안의 데이터를 가져올지

# 머신러닝 모델 준비
mindfulness_params = BrainFlowModelParams(BrainFlowMetrics.MINDFULNESS, BrainFlowClassifiers.DEFAULT_CLASSIFIER)
mindfulness = MLModel(mindfulness_params)
mindfulness.prepare()

# Muse 2 연결
board_shim.prepare_session()

# Muse 2 데이터 수집 시작
board_shim.start_stream()

# n초 + 1초 동안 데이터 수집
sleep(use_data_seconds + 1)

# 수집한 데이터를 변수에 저장
data = board_shim.get_current_board_data(num_points)

# 예측에 필요한 데이터 추출
bands = DataFilter.get_avg_band_powers(data, eeg_channels, sampling_rate, True)
feature_vector = bands[0]

# 집중도 예측
print(f"Mindfulness: {mindfulness.predict(feature_vector)[0]}")

# 모델 해제
mindfulness.release()

# Muse 2 데이터 수집 정지
board_shim.stop_stream()

# Muse 2 연결 해제
board_shim.release_session()

이번엔 아마도 저번보다는 쉽지 않을까 한다.. 코드적인걸로는..

이전 글에 대부분의 내용들을 넣어놓은거라서..

지금 새벽 4시 40분인데 다 쓰고 나면 짧겠지..? 라 생각 중..

 

참고로 이전 글에서 말했던 것 처럼 Mindfulness를 썼던게 Brainflow에서 Mindfulness로 썼어서 그걸로 쓴거다..

암튼 그렇다..

 

코드 설명 및 이해하기

설명할 수록 점점 코드 설명할게 적어져서 좋긴 하다.

전 글에서 한가지 안쓴게 있다면 필요한 라이브러리를 불러오는 것에 대해 안적었는데.. 뭐 간단한거니 여기서도 안적겠다!

여기서 적는건 이전에 설명하지 않았던 코드들만

 

mindfulness_params = BrainFlowModelParams(BrainFlowMetrics.MINDFULNESS, BrainFlowClassifiers.DEFAULT_CLASSIFIER)
mindfulness = MLModel(mindfulness_params)
mindfulness.prepare()

여기는 모델을 정의하는 부분이다. 어떤 모델을 쓸지 어떤 분류기를 쓸지 정할 수 있다.

코드에서는 모델과 분류기를 정하고 정한 파라미터로 모델을 할당해서 모델을 쓸 수 있도록 활성화 시키는 코드이다.

 

일단 Brainflow에선 기본적으로 어떤 모델과 분류기가 가능한지 아는 것이 좋다.

그러기 위해선 BrainFlowMetrics, BrainFlowClassifiers 정의 코드를 살펴봐야 한다.. 코드를 보면

# 코드 설명을 하기 위한 코드
class BrainFlowMetrics(enum.IntEnum):
    """Enum to store all supported metrics"""

    MINDFULNESS = 0  #:
    RESTFULNESS = 1  #:
    USER_DEFINED = 2  #:


class BrainFlowClassifiers(enum.IntEnum):
    """Enum to store all supported classifiers"""

    DEFAULT_CLASSIFIER = 0  #:
    DYN_LIB_CLASSIFIER = 1  #:
    ONNX_CLASSIFIER = 2  #:

이런식으로 정의가 되어있다. 

참고로 BrainFlowMetrics는 모델이 어떤 역할을 할지 선택하는 것이고, BrainFlowClassifiers는 그 역할을 계산할 수 있는 분류기를 선택할 수 있도록 하는 것이다.

 

각각에 대해서 설명하자면

BrainFlowMetrics 

• MINDFULNESS : 마음챙김, 집중도와 비슷할 순 있지만 좀 다른 것, 현재 자연스럽게 어느정도 집중 했는지
• RESTFULNESS : 휴식상태, 어느정도 휴식 상태인지
• USER_DEFINED : 사용자가 직접 정의

BrainFlowClassifiers

• DEFAULT_CLASSIFIER : 기본 분류기 (MINDFULNESS, RESTFULNESS 사용할 때)
• DYN_LIB_CLASSIFIER : .dll, .so 같은 C/C++로 만든 알고리즘을 적용하고 싶을 때
• ONNX_CLASSIFIER : ONNX 모델을 사용할 때 (USER_DEFINED 사용할 때)

여기에서 예시로 들 것은 Mindfulness이다. 설명을 적어놓기는 했지만 사실 나도 잘은 이해 못했다. 솔직히 저 설명으론 부족해서 추가로 말하자면, 내가 어느 순간에 공부나 무언가를 뚫어져라 보면서 갑자기 집중하는 것이 아닌.. 자연스럽게 무언가의 변화 등으로 자연스럽게 그것에 관심을 기울이는 것을 말하는 것 같다.

예를 들면 이전에서 한 Attention은 한 곳을 뚫어져라 바라볼 때 지표가 올라가고

Mindfulness는 밥 먹다가 수저가 바닥으로 떨어지며 그 소리에 기울인다 생각하면 된다.

수업 자료를 만들 때는 Mindfulness로 집중도를 구해서 게임할 수 있도록 했었는데.. 이젠 수정해야할 것 같다.. 하하하..

만약 이전 글에서 말한 수식으로 모델을 만든다면 집중하거나 안하는 데이터를 모두 수집한 후 수식으로 계산시키고 정규화 시킨 후에 학습시켜서 사용자 정의 모델로 등록하면 될 것 같긴 하다. (Brainflow에 등록 요청해볼까 생각중..)

 

기본 분류기는 아마도 Mindfulness나 Restfulness를 계산하는데만 쓰는 것 같다. 이에 대한건 predict 부분에서 더 자세하게 얘기할 것이다.

ONNX 분류기는 User_Defined로 선택했을 때 쓸 수 있는 것 같은데, 직접 Python으로 모델을 학습해서 만든 모델을 적용할 수 있도록 해주는 것이다. 이 분류기를 제작하는건 내가 반납 전에 쓸 분량을 다 쓰고 시간이 남는다면 써보고.. 안되면 아래에 링크한 튜토리얼을 보고 하면 될 것 같다.

 

추가로 여기서 한가지 생각이 들 것이다. 모델을 직접 정의할 수 있으면 BrainFlowMetrics는 왜 있는걸까? 라는 생각을 할 수 있다. 나도 제대로는 모르겠지만.. 내가 생각하기에는 Mindfulness랑 Restfulness를 쓸 수 있도록 하기 위해서 인 것 같다. 오픈소스다 보니 코드를 직접 수정할 수도 있는데 거기에서 추가적인 모델을 넣을 수 있도록 하거나 나중에 추가하기 위해서 저렇게 구분한 것 같다고 생각이 든다.

 

아무튼 여기에서는 Mindfulness를 계산하고 이 때문에 Default Classifier를 사용하게 된다.

 

# 예측에 필요한 데이터 추출
bands = DataFilter.get_avg_band_powers(data, eeg_channels, sampling_rate, True)
feature_vector = bands[0]

# 집중도 예측
print(f"Mindfulness: {mindfulness.predict(feature_vector)[0]}")

위에 2줄은 이전에 설명했기 때문에 패스하고

 

여기에서 볼 것은 저 아래 출력문이다.. 이전에 할당한 Mindfulness 모델에 수집한 데이터의 평균 파워를 넣어 예측하게 하는 코드이다.

즉 이전 글에서 설명한 평균의 Delta, Theta, Alpha, Beta, Gamma 값을 넣어 예측시키는 코드라고 보면 된다.

 

그런데 또 의문이 들것이다. 왜 예측한 것의 0번 인덱스 값을 가져오라고 하는지.. 일단 확인해보니 0번 인덱스 외의 값은 없다. 그런데 리스트로 반환하는 이유가 뭔지 생각해보니.. 아마도 다른 사용자 정의 분류기를 사용하게 되면 여러 값을 리턴해야하는 경우가 있을테니 그것 때문인 듯 하다.

아무튼 리스트로 반환되기 때문에 [값] 이런식으로 출력되지 않게 하기 위해 0 인덱스로 접근해서 데이터를 가져와 출력하게 되어있다.

 

참고로 찾으면서 방금 발견한게 있는데.. 수업 자료를 만들 때는 저거를 1초나 2초로 했었다..

그런데 갑자기 최소 몇초를 해야하는 생각이 들어서 더 찾아봤는데.. 당시에는 못 찾은 주석이 있었다..

# recommended window size for eeg metric calculation is at least 4 seconds, bigger is better

음.. EEG로 어떤 것을 계산해야한다면 4초 이상의 데이터는 필요하다고.. 한다.. 클 수록 좋다고 하고.. 흐음.. 

수업 자료를 고쳐서 다시 교수님께 드려야 할 것 같다는 생각이 든다.. 하하;;

 

그럼 Mindfulness의 계산 방법에 대해 궁금한점이 생길 것이다.

위에서 말했듯이 머신러닝이다. 하지만 더 자세히 보기 위해 코드로 보겠다!

# 코드 설명을 하기 위한 코드
# src/ml/generated/mindfulness_model.cpp

const double mindfulness_coefficients[5] = {-1.4060899708538128,2.597693987367105,-30.96470526503066,12.04593986553724,45.773017975354556};
double mindfulness_intercept = 0.000000;

# 코드 설명을 하기 위한 코드
# src/ml/mindfulness_classifier.cpp

int MindfulnessClassifier::predict (double *data, int data_len, double *output, int *output_len)
{
    if ((data_len < 5) || (data == NULL) || (output == NULL))
    {
        safe_logger (spdlog::level::err,
            "Incorrect arguments. Null pointers or invalid feature vector size.");
        return (int)BrainFlowExitCodes::INVALID_ARGUMENTS_ERROR;
    }
    double value = 0.0;
    for (int i = 0; i < std::min (data_len, 5); i++)
    {
        value += mindfulness_coefficients[i] * data[i];
    }
    double mindfulness = 1.0 / (1.0 + exp (-1.0 * (mindfulness_intercept + value)));
    *output = mindfulness;
    *output_len = 1;
    return (int)BrainFlowExitCodes::STATUS_OK;
}

이게 Mindfulness를 계산하는 코드이다. Python이 아닌 이유는 Brainflow는 C++로 만든걸 Python에 이식한 것이기 때문이다.. 저건 메인 C++에 있는 코드..

 

하나씩 보자면 일단 mindfulness_coefficients, mindfulness_intercept 이거는 인공지능을 어느정도 했었다면 알 수 있다.

가중치와 편향 값이다. mindfulness_coefficients가 가중치 값, mindfulness_intercept 편향 값..

5개인 이유는 입력 값인 Delta, Theta, Alpha, Beta, Gamma를 입력 값으로 계산하기 때문..

 

그러면 예측 코드를 하나씩 봐보자!

아래에서 설명할껀 퍼셉트론 대로 계산한다고 생각하면 된다.

퍼셉트론은 입력 값과 가중치 값을 각각 곱하고 만나는 노드에서 모두 더해 편향 값을 더한 후 활성 함수를 거치는거라 생각하면 된다. (말로는 이해하기 어렵고 모르면 퍼셉트론에 대해서 공부하고 오면 될 것 같다.)

# 코드 설명을 하기 위한 코드
double value = 0.0;
for (int i = 0; i < std::min (data_len, 5); i++)
{
    value += mindfulness_coefficients[i] * data[i];
}

 

 

그 중 위 코드는 입력값과 가중치를 각각 곱한 후 모두 더하는 코드라 보면 된다.

# 코드 설명을 하기 위한 코드
double mindfulness = 1.0 / (1.0 + exp (-1.0 * (mindfulness_intercept + value)));

이건 위에서 곱하고 더한 값인 value에 편향 값을 더하고 활성화 함수를 거치는 것이다. 

그리고 활성화 함수는 시그모이드 함수를 사용하는데 f(x) = 1 / (1 + e^(-x)) 이런 수식이다. 솔직히 몰라도 된다.

 

그리고 실제 계산 과정으로 예를 보여주겠다!

먼저 계산하기 위해서 입력 값을 정의하면..

• Delta : 0.53311590
• Theta : 0.30351431
• Alpha : 0.07848158
• Beta : 0.06465923
• Gamma : 0.02022897

이렇다. 이건 실제로 내가 방금 착용해서 얻은 값이다. 참고로 미리 내부에서 계산된 것은 0.33481021 이 값이다.

 

아무튼 계산해보면

일단 처음에 입력 값과 가중치를 곱한다.

• Value0 : Delta(0.53311590) * Weight0(-1.4060899708538128) = -0.7496089202927042
• Value1 : Theta(0.30351431) * Weight1(2.597693987367105) = 0.7884372981668756
• Value2 : Alpha(0.07848158) * Weight2(-30.96470526503066) = -2.430158993433925
• Value3 : Beta(0.06465923) * Weight3(12.04593986553724) = 0.7788811963319415
• Value4 : Gamma(0.02022897) * Weight4(45.773017975354556) = 0.925941007432908

그 후에는 곱해진 모두를 더한다.

• Value : Value0(-0.7496089202927042) + Value1(0.7884372981668756) + Value2(-2.430158993433925) + Value3(0.7788811963319415) + Value4(0.925941007432908) = -0.6865084117949039

그 후 편향 값을 더한다.

• Value : Value(-0.6865084117949039) + Bias(0.000000) = -0.6865084117949039

참고로 편향 값이 0이기에 값의 변화는 없다.

그 후 활성화 함수를 거친다.

• Result : 1.0 / (1.0 + math.exp(-1.0 * Value)) = 0.3348102440015052

이런 과정으로 계산되게 된다. 결론적으로 내부적으로 계산된 값인 0.33481021과 비교하면 반올림되서 잘린거 빼고는 결과값의 차이는 없다. 그렇다는건 위 계산 과정이 맞다는 것이 되니 저걸 보면 된다!

 

아무튼 계산 과정이 이렇다. 어떻게 보면 간단할 수도 있고 복잡할 수도 있다.

참고로 DNN(깊은 신경망)은 저런 계산하는 것이 겁나게 많은 것 뿐이다.

 

당시 수업 자료 만들 때의 PPT에 내가 만들어둔 사진이다. 아래 사진을 보면 이해가 더 쉬울지도 몰라서 올려놓겠다.

지금 봤는데.. 가중치 값이 업데이트 되었나보다.. 사진에 있는건 대략 5개월 전의 가중치 값..

 

참고로 재미있는점이 한가지 있는데.. Restfulness를 계산하는 과정이다..

예전에도 찾았었던 것 같은데.. 지금봐도 어이가 없다.

# 코드 설명을 하기 위한 코드
# src/ml/inc/restfulness_classifier.h

#pragma once

#include "brainflow_constants.h"
#include "mindfulness_classifier.h"


class RestfulnessClassifier : public MindfulnessClassifier
{
public:
    RestfulnessClassifier (struct BrainFlowModelParams params) : MindfulnessClassifier (params)
    {
    }

    int predict (double *data, int data_len, double *output, int *output_len)
    {
        int res = MindfulnessClassifier::predict (data, data_len, output, output_len);
        if (res != (int)BrainFlowExitCodes::STATUS_OK)
        {
            return res;
        }
        *output = 1.0 - (*output);
        return res;
    }
};

안믿을 것 같아서 Restfulness의 전체 코드를 가져왔다. 

해석해보면 MindfulnessClassifier를 할당 받아서 MindfulnessClassifier에 있는 예측하는 코드로 예측시킨 후

그 값을 (1.0 - 값) 이걸로 계산한다는 것이다.

즉 (1.0 - Mindfulness 예측 값)이다. 이 말은 Restfulness는 Mindfulness의 반대 값이라고 생각하면 될 것 같다..

그래서 더 해보면 1이 나오는걸 알 수 있다.

 

암튼 저런 계산 과정으로 마음챙김도가 나오는 것이다.

저런 마음 챙김 모델도 학습하는 방법도 나와있는데 그건 맨 아래 링크에 달아두었다.

 

mindfulness.release()

위에서 할당한 모델을 해제하는 것이다.

 

암튼 이제 설명을 다 했으니 코드를 실행해봐야 한다. (위에서 이미 실행한 결과는 무시하시길..)

 

코드 실행

[2025-02-21 09:47:34.717] [board_logger] [info] incoming json: {
    "file": "",
    "file_anc": "",
    "file_aux": "",
    "ip_address": "",
    "ip_address_anc": "",
    "ip_address_aux": "",
    "ip_port": 0,
    "ip_port_anc": 0,
    "ip_port_aux": 0,
    "ip_protocol": 0,
    "mac_address": "",
    "master_board": -100,
    "other_info": "",
    "serial_number": "Muse-0465",
    "serial_port": "",
    "timeout": 0
}
[2025-02-21 09:47:34.717] [board_logger] [info] Use timeout for discovery: 6
[INFO] SimpleBLE: D:\a\brainflow\brainflow\third_party\SimpleBLE\simpleble\src\backends\windows\Utils.cpp:33 in initialize_winrt: CoGetApartmentType: cotype=-1, qualifier=0, result=800401F0
[INFO] SimpleBLE: D:\a\brainflow\brainflow\third_party\SimpleBLE\simpleble\src\backends\windows\Utils.cpp:41 in initialize_winrt: RoInitialize: result=0
[2025-02-21 09:47:34.764] [board_logger] [info] found 1 BLE adapter(s)
[2025-02-21 09:47:34.958] [board_logger] [info] Found Muse device
[2025-02-21 09:47:36.400] [board_logger] [info] Connected to Muse Device
[2025-02-21 09:47:37.405] [board_logger] [info] found control characteristic
Mindfulness: 0.97519466

이런식으로 뜬다. 로그 정보들은 이전에도 설명한 것과 같이 똑같으니 무시하고

 

"Mindfulness: 0.97519466" 값만 보면 된다.

다만 전에 이 집중도하고 전 글의 집중도하고 다르긴 하지만 비슷하기 때문에 자세하게 측정결과는 작성 안할 것이다.

어차피.. 내가 설명한 여기에서의 Mindfulness 집중도는.. 나도 잘 만들어내기가 어렵다..

 

참고로 Restfulness는 Mindfulness를 전부 Restfulness로 바꾸면 된다.

 

그럼으로 이번 글은 여기에서 끝내는걸로..

 

추가로 이전 글에서도 올린 것 처럼 실시간으로 돌릴 수 있는 코드!

import keyboard
from time import sleep
from brainflow.board_shim import BoardShim, BrainFlowInputParams, BoardIds
from brainflow.data_filter import DataFilter
from brainflow.ml_model import MLModel, BrainFlowModelParams, BrainFlowMetrics, BrainFlowClassifiers

board_id = BoardIds.MUSE_2_BOARD
params = BrainFlowInputParams()
params.serial_number = "Muse-0465" # Muse 2의 고유 시리얼 넘버
board_shim = BoardShim(board_id, params)

use_data_seconds = 5 # n초 동안 데이터 수집
sampling_rate = BoardShim.get_sampling_rate(board_id)
eeg_channels = BoardShim.get_eeg_channels(board_id)
num_points = use_data_seconds * sampling_rate # 몇 초 동안의 데이터를 가져올지

# 머신러닝 모델 준비
mindfulness_params = BrainFlowModelParams(BrainFlowMetrics.MINDFULNESS, BrainFlowClassifiers.DEFAULT_CLASSIFIER)
mindfulness = MLModel(mindfulness_params)
mindfulness.prepare()

# Muse 2 연결
board_shim.prepare_session()

# Muse 2 데이터 수집 시작
board_shim.start_stream()

# n초 + 1초 동안 데이터 수집 - 안하면 오류남
sleep(use_data_seconds + 1)

while True:
    # 나가기 조건
    if keyboard.is_pressed('q'):
        break
   
    # 수집한 데이터를 변수에 저장
    data = board_shim.get_current_board_data(num_points)
    
    # 예측에 필요한 데이터 추출
    bands = DataFilter.get_avg_band_powers(data, eeg_channels, sampling_rate, True)
    feature_vector = bands[0]

    # 집중도 예측
    print(f"Mindfulness: {mindfulness.predict(feature_vector)[0]}")

# 모델 해제
mindfulness.release()

# Muse 2 데이터 수집 정지
board_shim.stop_stream()

# Muse 2 연결 해제
board_shim.release_session()

 

이제 올릴려 한 것들은 다 올렸고.. 추가로 나머지는 인공지능을 만들어서 적용해보거나 게임을 하면서 집중도는 어떤지 측정해서 올려볼려고 한다.

 

암튼 짧게 끝냈다! 다행히 이번엔 쓰는 시간이 4시간 정도 걸렸다.

 

위에서 언급한 내용들의 관련 링크

Tensorflow로 모델을 제작하여 적용

https://brainflow.org/2022-09-08-onnx-tf/

BrainFlow with onnxruntime (TensorFlow edition)

Scikit Learn으로 모델을 제작하여 적용

https://brainflow.org/2022-06-09-onnx/

BrainFlow with onnxruntime

BrainFlow 학습 코드

https://github.com/brainflow-dev/brainflow/blob/master/src/ml/train/train_classifiers.py

brainflow/src/ml/train/train_classifiers.py at master · brainflow-dev/brainflow