IMU 센서 데이터, 실시간 데이터 그래프 시각화 with Python

미리보는 키포인트 기술들

• 실시간 데이터 그래프 시각화
• 실시간 통신에 따른 그래프 딜레이 감소 기법
• 여러 그래프 사용

 

소개

 

IMU 센서의 데이터를 시리얼 통신으로 받아 실시간으로 그래프를 그려주는 코드입니다.

IMU 센서란?
IMU(Inertial Measurement Unit) 센서는 관성 측정 장비로, 가속도, 자이로스코프, 지자기 센서를 포함하는 단일 센서 장치입니다. 각도, 가속도, 속도 및 위치에 대한 정보를 제공하기 위해 사용되며, 드론, 비행 장치, 스마트폰 등 많은 곳에 사용되고 있습니다. 포함된 센서들로부터 데이터를 수집, 분석하여 자세 정보와 위치 정보를 추정할 수 있습니다.

가속도, 자이로스코프, 지자기 센서는 각각 3개의 축을 가져 총 9개의 축을 가집니다. 9개의 각기 다른 숫자 나열만을 출력해서 보는 것만으로는, 각 축이 어떻게 센싱하고 어떠한 특징을 가졌는지 파악하기에 어려움이 있어 만들었습니다.

 

깃허브 링크

GitHub - 0x42-Embedded-Robotics-club/IMU-DataVisualization

 

구현

시리얼 통신 및 문자열 split을 통한 데이터 추출

다양한 방법이 있겠지만 여기서는 개발 보드에서 IMU 센서의 데이터를 받아오기 위해 시리얼 통신을 사용합니다.

먼저 시리얼 통신을 사용하기 위해선 pyserial 패키지를 설치해야 합니다.

pip install pyserial

시리얼 통신을 받아 사용하는 코드는 다음과 같습니다.

# pyserial 임포트
import serial

# 시리얼 통신 포트와 baud rate 설정
ser = serial.Serial("/dev/cu.usbmodem1103", 115200) 

# 시리얼 통신 읽기
serialData = ser.readline().decode()

# 문자열 split을 통하여 필요 데이터 추출
accelX, accelY, accelZ, gyroX, gyroY, gyroZ, magX, magY, magZ = serialData.split(',')

시리얼 통신을 위해선 serial.Serial(...)을 사용하여 (인자 1)어느 통신 포트와 (인자 2)어떤 baud rate를 사용하여 통신을 주고받을 것인지 설정해주어야 합니다.
시리얼 포트명은 주로 운영체제마다 일정한 네이밍 규칙이 있습니다. 맥에서는 "/dev/cu.usbmodem..."인 경우가 많으며, 윈도우에서는 "COM..."과 같은 형식을 많이 볼 수 있습니다. 만약 아두이노 IDE가 설치되어 있다면 쉽게 확인할 수 있습니다. 아두이노 보드이든 아니든간에 아두이노 IDE는 웬만해선 연결된 모든 장치를 잡아줍니다. 거기서 시리얼 포트명을 확인할 수도 있어요.

이후 보낸 보드에서 보낸 데이터는 ser.readline().decode()함수를 사용하여 문자열로 받아볼 수 있습니다.
이렇게 string으로 받은 데이터는 필요에 따라서 파싱해주면 됩니다. 여기서는 %d,%d,%d,...,%d 와 같은 규칙으로 보내주고 있기에 문자열의 split(',')함수를 사용하여 쉼표를 기준으로 나눠주었습니다.

 

실시간 데이터 그래프 시각화

Matplotlib는 파이썬에서 과학 연산 및 데이터 시각화를 위한 라이브러리입니다. 위에서와 마찬가지로 pip 명령어를 통해 설치해주세요.

pip install matplotlib

실시간으로 그래프를 그려주기 위한 시각화 관련 코드는 다음과 같습니다.

# 그래프에서 보여줄 최대 정점
max_points = 100

# 그래프 설정
fig = plt.figure()
plt.subplots_adjust(left=0.125, bottom=0.1, right=0.9, top=0.9, wspace=0.2, hspace=0.5)

accelAxes = fig.add_subplot(1, 1, 1, xlim=(0, max_points), ylim=(-3.0, 3.0))

# accelAxes 그래프 상세 설정
# 그래프 제목 설정
accelAxes.set_title('Accelation')
# x축 설정
accelLineX, = accelAxes.plot([], [], lw=2)
accelLineX, = accelAxes.plot(np.arange(max_points), np.ones(max_points, dtype=np.float64)*np.nan, lw=2, color='red', label='x')
# y축 설정
accelLineY, = accelAxes.plot([], [], lw=2)
accelLineY, = accelAxes.plot(np.arange(max_points), np.ones(max_points, dtype=np.float64)*np.nan, lw=2, color='green', label='y')
# z축 설정
accelLineZ, = accelAxes.plot([], [], lw=2)
accelLineZ, = accelAxes.plot(np.arange(max_points), np.ones(max_points, dtype=np.float64)*np.nan, lw=2, color='blue', label='z')
# 그래프 옆에 x, y, z축 안내 정보 표기
accelAxes.legend(loc='upper left', bbox_to_anchor=(1.0, 1.0))

# 그래프 필요 값 초기화 함수
def init():
    return accelLineX, accelLineY, accelLineZ

# 그래프 업데이트 함수
def animate(i):
    accelX, accelY, accelZ = getAllSensorDataOnQueue()
    if (accelX.count == 0 || accelY.count == 0 || accelZ.count == 0):
        return

    # accel
    oldAccelX = accelLineX.get_ydata()
    newAccelX = np.r_[oldAccelX[1:], accelX]
    accelLineX.set_ydata(newAccelX)

    oldAccelY = accelLineY.get_ydata()
    newAccelY = np.r_[oldAccelY[1:], accelY]
    accelLineY.set_ydata(newAccelY)

    oldAccelZ = accelLineZ.get_ydata()
    newAccelZ = np.r_[oldAccelZ[1:], accelZ]
    accelLineZ.set_ydata(newAccelZ)

    # gyro
    # (생략)

    # mag
    # (생략)

    return accelLineX, accelLineY, accelLineZ

# 그래프 업데이트 등록 및 시작
anim = animation.FuncAnimation(fig, animate, init_func=init, frames=200, interval=20, blit=False)
plt.show()

 

코드 해설

# 그래프에서 보여줄 최대 정점
max_points = 100

max_points는 그래프에서 보여줄 최대 정점을 의미합니다. 즉 그래프의 x축(가로)의 길이를 의미합니다. 여기서 그릴 그래프는 '실시간' 데이터를 쉽게 파악하기 위함으로 일정 시간이 지난 데이터는 그래프에서 사라집니다.

= 좀 더 오랜 기간을 띄워주고 싶다면 이 변수를 늘리면 됩니다.

# 그래프 설정
fig = plt.figure()
plt.subplots_adjust(left=0.125, bottom=0.1, right=0.9, top=0.9, wspace=0.2, hspace=0.5)

accelAxes = fig.add_subplot(1, 1, 1, xlim=(0, max_points), ylim=(-3.0, 3.0))

matplotlib의 figure 객체를 생성하여 그래프가 그려질 공간과 전체적인 항목을 설정합니다. subplots_adjust 메서드를 사용하여 그래프의 위치와 간격 등의 설정을 할 수 있습니다.
그래프 객체인 accelAxes를 생성하고 설정을 진행합니다. add_subplot을 사용하여 figure 객체에 그래프를 추가하면 됩니다. 여기서 첫 3개 인자는 다수의 그래프를 사용하는 내용에서 다시 다루겠습니다. 마지막 인자로 들어간 ylim = (-3.0, 3.0)은 그래프의 y축의 범위를 지정합니다.

# accelAxes 그래프 상세 설정
# 그래프 제목 설정
accelAxes.set_title('Accelation')
# x축 설정
accelLineX, = accelAxes.plot([], [], lw=2)
accelLineX, = accelAxes.plot(np.arange(max_points), np.ones(max_points, dtype=np.float64)*np.nan, lw=2, color='red', label='x')
# y축 설정
accelLineY, = accelAxes.plot([], [], lw=2)
accelLineY, = accelAxes.plot(np.arange(max_points), np.ones(max_points, dtype=np.float64)*np.nan, lw=2, color='green', label='y')
# z축 설정
accelLineZ, = accelAxes.plot([], [], lw=2)
accelLineZ, = accelAxes.plot(np.arange(max_points), np.ones(max_points, dtype=np.float64)*np.nan, lw=2, color='blue', label='z')
# 그래프 옆에 x, y, z축 안내 정보 표기
accelAxes.legend(loc='upper left', bbox_to_anchor=(1.0, 1.0))

여기서는 위에서 생성한 accelAxes 그래프 객체에 대한 상세 설정을 합니다.
가속도 센서는 x, y, z 3개 축을 가지므로 각 축을 생성하고 설정합니다. 마지막의 legend() 메서드를 호출해서 그래프의 원하는 위치에 x, y, z 축에 대한 안내 정보를 표기할 수 있습니다.

# 그래프 필요 값 초기화 함수
def init():
    return accelLineX, accelLineY, accelLineZ

init()함수는 실시간으로 그래프를 그려줄 때 필요한 FuncAnimation에서 처음 실행될 때 호출되는 함수입니다. 이 함수는 그래프에 표기할 데이터를 초기화하고, 어떤 데이터가 표시될지 정합니다.

# 그래프 업데이트 함수
def animate(i):
    accelX, accelY, accelZ = getAllSensorDataOnQueue()
    # 예외 처리
    if (accelX.count == 0 || accelY.count == 0 || accelZ.count == 0):
        return

    # accel
    oldAccelX = accelLineX.get_ydata()
    newAccelX = np.r_[oldAccelX[1:], accelX]
    accelLineX.set_ydata(newAccelX)

    oldAccelY = accelLineY.get_ydata()
    newAccelY = np.r_[oldAccelY[1:], accelY]
    accelLineY.set_ydata(newAccelY)

    oldAccelZ = accelLineZ.get_ydata()
    newAccelZ = np.r_[oldAccelZ[1:], accelZ]
    accelLineZ.set_ydata(newAccelZ)

    # gyro
    # (생략)

    # mag
    # (생략)

    return accelLineX, accelLineY, accelLineZ

animate()함수는 FuncAnimation 클래스의 frames 속성에서 정의된 횟수만큼 반복적으로 실행되어 그래프의 내용을 업데이트하는 함수입니다.
accelX, accelY, accelZ = getAllSensorDataOnQueue()을 수행하여 그래프에 추가할 값의 배열들을 받아옵니다. 여기서 배열을 받는 이유는 아래 그래프 딜레이 감소 기법과 관련되어 있습니다.
각 축에 대해 get_ydata() 메서드를 호출하여 기존 데이터를 가져온 뒤, rp.r() 함수를 호출하여 새로운 데이터와 결합하고 set_ydata()를 통해 새로운 값을 넣어주면 됩니다.

# 그래프 업데이트 등록 및 시작
anim = animation.FuncAnimation(fig, animate, init_func=init, frames=200, interval=20, blit=False)
plt.show()

FuncAnimation은 Matplotlib의 animation 모듈에서 제공하는 클래스로, 그래프를 애니메이션으로 구성할 때 사용합니다. 위에서 만들어둔 figure 객체와 init, animate 함수를 전달해주어야 합니다. 여기서 frames, interval, blit은 차례로 애니메이션 프레임 수, 시간 간격, 블릿 기법 적용 여부를 의미합니다.
마지막으로 plt.show()를 호출하여 그래프를 표시합니다.

 

실시간 통신에 따른 그래프 딜레이 감소 기법

위 코드에서 animate 함수에서 새로 그려줄 값을 배열의 형태로 받아 처리하였습니다. 여기서는 왜 그래야 하는지 알아보겠습니다.
처음에 데이터를 처리할 땐 animate() 함수에서 시리얼 통신을 받아 처리해주고 있었습니다. 이와 같이 처리할 경우 실시간 처리되어야 할 그래프의 반응이 10초 이상 지연되는 문제가 발생합니다. 원인은 다음과 같이 한 줄로 정리할 수 있습니다.

• 그래프를 업데이트하는 속도보다, 개발 보드에서 데이터를 전송하는 속도가 훨씬 빠르다.

이 문제는 쓰레드와 일괄처리로 해결할 수 있습니다.
자세한 해법은 코드와 함께 보죠.

import queue
import threading
import numpy as np

# 센서 데이터 큐
sensorDataQueue = queue.Queue()

# 센서 값 읽기 쓰래드 함수
def serialSensorRead():
    while True:
        serialData = ser.readline().decode()

        accelX, accelY, accelZ = serialData.split(',')

        accelX = float(accelX)
        accelY = float(accelY)
        accelZ = float(accelZ)

                # 읽어들인 값을 큐에 삽입
        sensorDataQueue.put((accelX, accelY, accelZ))

...(생략)...

# 큐에 쌓인 데이터 통합하여 반환. 그래프에 한번에 적용하여 속도 증대.
def getAllSensorDataOnQueue():
    accelXBunch = []
    accelYBunch = []
    accelZBunch = []
    while not sensorDataQueue.empty():
        accelX, accelY, accelZ, gyroX = sensorDataQueue.get_nowait()
        accelXBunch.append(accelX)
        accelYBunch.append(accelY)
        accelZBunch.append(accelZ)
    return tuple(accelXBunch), tuple(accelYBunch), tuple(accelZBunch)

# 시리얼 통신 센서값 읽기 함수 쓰래드 등록 및 시작
serialSensorReadThread = threading.Thread(target=serialSensorRead, daemon=True)
serialSensorReadThread.start()

위 코드의 기능을 정리해 보면 다음과 같습니다.

• 쓰레드에서 계속해서 시리얼 통신을 통해 데이터를 읽어들여, 그 데이터를 큐에 저장.
• 큐에 저장된 데이터를 통합하여 반환하여 그래프 업데이트에서 일괄 처리를 지원함.

serialSensorRead 함수에서는 계속해서 시리얼 통신을 통해 데이터를 읽어들여, 그 데이터를 큐에 저장합니다.

getAllSensorDataOnQueue 함수는 큐에 저장된 센서 데이터를 통합하여 반환하여 그래프에서 그동안의 데이터를 일괄 처리할 수 있도록 합니다. 큐가 빌 때까지 계속해서 데이터를 가져오는 것을 보장하기 위해서 get_nowait() 메서드를 사용하고 있습니다.

serialSensorReadThread는 위에서 구현한 serialSensorRead 함수를 대상으로 하는 쓰레드를 생성하고, start() 메서드를 호출하여 쓰레드를 시작합니다. daemon=True 옵션은 쓰레드가 메인 프로세스가 종료될 때 같이 종료되도록 설정합니다.

 

여러 그래프 사용

IMU 센서는 가속도뿐만이 아닌, 자이로스코프, 지자기 센서를 가지고 있습니다. 더 나아가 자세 추정치와 같은 데이터도 시각화해줄 필요가 있습니다.

여러 그래프를 사용하기 위한 코드는 아래와 같습니다.

accelAxes = fig.add_subplot(3, 1, 1, xlim=(0, max_points), ylim=(-3.0, 3.0))
gyroAxes = fig.add_subplot(3, 1, 2, xlim=(0, max_points), ylim=(-300.0, 300.0))
magAxes = fig.add_subplot(3, 1, 3, xlim=(0, max_points), ylim=(-200.0, 200.0))

fig.add_subplot() 함수를 호출하여 그래프를 추가할 수 있습니다.
첫 번째 인자는 그래프의 행(row) 개수, 두 번째 인자는 열(column) 개수를, 세 번째 인자는 해당 그래프의 인덱스를 의미합니다.
즉, 위의 코드에서는 사진과 같이 3 개의 그래프(accelAxes, gyroAxes, magAxes)를 1행 3열로 나열하여 그려지게 됩니다.

 

ISSUE

그래프 데이터의 shape 불일치 문제가 있습니다. 개선되는 데로 게시글을 수정하겠습니다.
혹시 해결 방법을 알고 계시는 분께선 공유를 부탁드립니다. 감사합니다!

 

해당 깃허브 이슈

:bug: 그래프 데이터 shape 불일치 · Issue #3 · 0x42-Embedded-Robotics-club/IMU-DataVisualization

 

 

방문해 주셔서 감사합니다.
도움이 되었으면 좋겠습니다.