[Monitoring System] 3. UDP 통신으로 실시간 진동 데이터 Plotting(feat. matplotlib)

이번 포스팅에서는 이전 mqtt에서 사용한 예제 데이터보다 2만 배 정도 많은 데이터를 DAQ에서 받아오는 UDP 예제를 만들 것이다. 구현하려는 웹 서비스에 온도 데이터 말고 진동 데이터도 넣어보고 싶은데, 그것을 위해서 우선 대용량 데이터이기 때문에 복잡해지는 데이터베이스나 서버를 제외하고 소켓 통신한 결과를 matplotlib로 로컬에 간단하게 띄워보도록 하겠다.

오늘의 목표

1. UDP 통신

UDP는 소켓 통신의 일종으로, 다른 소켓 통신방법인 TCP와는 다르게 연결을 보장해주지 않는다. 나쁘게 들릴 수도 있겠으나 client와 server의 연결을 요청하고, 승인하고 연결한 후 통신을 시작하는 TCP와 달리 데이터를 매번 보낼 때마다 데이터의 목적지(ip, port번호)를 함께 실어서 일단 냅다 보내는 방식이다. 그러니 데이터를 잘 받았는지 검증도 해주지 않는다. 연결이 지속적이지 않기 때문이다.

장점은 데이터를 받지 못한 경우에는 재시도 해주는 TCP와 달리 보내면 끝이기 때문에 데이터 양이 어마무시하게 방대하고 실시간으로 왕창 보내야하며, 몇 개 잃어버려도 크게 상관없는 경우에는 UDP가 빠르고 좋을 수 있다. 바로 지금! 초당 2만여 개의 데이터가 발생하는 진동 센서와 같은 경우가 그렇다.

진동 데이터 관련 이론은 ...아직 안 썼구나(깨달음) 조만간 올려서 링크를 걸도록 하겠다. 사실 아직 잘 모르겠다^^; 각설하고 코드를 보자.

Python 예제

- Server.py

import socket

server = ('YOUR_IP_ADDRESS', 2001)
serverSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
serverSocket.bind(server) 

data = serverSocket.recv(1024)

- Client.py

import socket

server = ('YOUR_IP_ADDRESS', 2001)
clientSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

data = dummy_data
clientSocket.sendto(data, server)

간단하다. YOUR_IP_ADDRESS 에 server(데이터 받는 쪽)의 ip를 넣고, client에서 데이터를 보내고 server에서 데이터를 읽어들인다. client코드의 sendto부분을 보면 ip와 port번호를 함께 전송하는 것을 볼 수 있다.

주의해야할 점은, server에서 데이터를 받을 때 정해둔 버퍼사이즈보다 데이터의 크기가 크면 안된다. 현재 예제에서는 1024 byte로 해두었으니, 한 번에 받는 데이터가 이 사이즈를 넘어가서는 안된다는 것이다.

2. 데이터 송수신

여기서부터는 각자 가지고 있는 DAQ 장치의 sampling rate 등등 여러 하드웨어적인 스펙에 따라 달라질 수 있다. 만일 dummy data로 한다면 5kHz 정도로 생성해도 좋을 것이다. 초당 5000개 정도의 데이터이다.

현재 사용하고 있는 DAQ 장치는 라즈베리 파이 위에 장착하는 형태로 된 mcc172 DAQ hat이다. 초당 약 18000개의 데이터를 샘플링하도록 설정하였다.

데이터의 sampling rate는 그렇지만 실제로 데이터가 1초에 한 번 들어오지는 않는다. 이 예제의 경우 한 번에 약 2045개 가량의 데이터를 전송하고, 수신한다. 데이터의 양이 많으면 float array 형태보다는 byte로 packing해서 보내고, 받아서 unpack하는 것이 더 용이하기 때문에 그렇게 코드를 작성하였다.

- Client.py

import socket, struct

server = ('YOUR_IP_ADDRESS', 2001)
clientSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

...

data = struct.pack('%sf' %len(signal), *signal)
socket.sendto(data, server)

clientSocket.sendto(data, server)

float array를 byte화하기 위해서 struct 라이브러리를 사용하였다.

struct.pack('%sf' %len(signal), *signal)

이는 signal의 길이와 같은 개수의 float 값들을 byte화한다는 뜻이다.

- Server.py

import socket, array

server = ('YOUR_IP_ADDRESS', 2001)
serverSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
serverSocket.bind(server) 

data = serverSocket.recv(8192)
signals = array.array('f')
signals.frombytes(data)

서버 쪽에서는 버퍼 사이즈를 정해주어야한다. 보내는 데이터가 최대 2045개 float 데이터이므로 4byte씩 곱하고 여유를 잡아 8192로 설정하였다. 내 경우에는 데이터의 양이 매번 일정하지 않아 struct의 unpack을 활용해서 unpacking할 수 없었다.

그런데 윈도우 운영체제가 server인 경우, 이렇게 했을 때 버퍼 사이즈 오버 에러가 발생했다. 리눅스-리눅스에서는 문제가 없었는데, 이유는 잘 모르겠다. 이유를 아시는 분은 댓글로.......

그리고 array 라이브러리를 사용하여 받은 데이터를 float 단위로 잘라주고, byte를 다시 변환하면 된다.

3. Plotting

무한루프로 데이터를 받아오다보면 프로세스가 데이터를 왕창 처리하느라 바빠서 그래프까지 그릴 여유가 없다. 이를 해결하기 위해 multi-thread를 활용했다. 데이터를 받아서 갱신하는 쪽을 thread로 분기시키고, 메인 thread에서 그래프를 그린다.

3.1 thread 생성하기

import socket, array
from threading import Thread

def run() -> None:
    print("Connected", flush=True)

    while True:
        data = serverSocket.recv(8192)
        signals = array.array('f')
        signals.frombytes(data)
        entire_data.extend(signals)

if __name__ == "__main__":

    # define socket setting and bind
    server = ('YOUR_IP_ADDRESS', 2001)
    serverSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    serverSocket.bind(server) 

    entire_data = [0]

    # start UDP in another thread
    t = Thread(target=run, args=())
    t.start()

먼저 thread를 만들었다. entire_data에 전체 데이터를 지속적으로 저장하도록 해주었다.

3.2 real time plotting - matplotlib.animation

plotting은 matplotlib의 animation을 활용해서 그리기로 하였다.

https://matplotlib.org/stable/api/animation_api.html

matplotlib.animation — Matplotlib 3.7.1 documentation

import matplotlib.pyplot as plt
import matplotlib.animation as animation

class RealTimeAnimation:
    """
    Animate original voltage signal graph and fft result.
    """
    def __init__(self):
        # define plot
        self.fig = plt.figure(figsize=(16, 6))
        self.original = self.fig.add_subplot(1, 1, 1)

        # connect animation method
        self.animation = animation.FuncAnimation(self.fig, self.update, interval=100)

    def update(self, i) -> None:
        """
        Run every interval and update graphs.
        Using lock.

        Args:
            i : interval
        """

        # update original graph
        if len(entire_data) > 110000:
            graph_data = entire_data[len(entire_data) - 100000:]
        else: 
            graph_data = entire_data

        self.original.clear()
        self.original.plot(graph_data)

클래스를 생성해서 animation을 정의해주었다. FuncAnimation에는 plot의 figure, 그리고 update function, interval이 주어져야한다. 차례로 animation을 그릴 공간, interval마다 실행될 함수이다.

update 함수를 보면 animation을 전체 다 그리는 게 아니라 100000개만 그리도록 잘라주었다. 초당 2만여 개의 데이터가 들어오기 때문에 약 5초 단위로 보여주는 것이다. 너무 자주 불려서 속도를 느리게 할 것을 염려하여 11만개가 넘으면 자르는 것으로 해주었다.

이 정도 되면 이제 슬슬 뭔가 하나 빠진 기분이 든다. 바로 entire_data 의 접근 문제이다. entire_data는 소켓 통신을 하는 thread에서 지속적으로 업데이트를 해주고 있는데, 동시에 animation에서 읽어들여서 자르거나 복사한다. 두 thread 모두 실시간으로 빠르게 실행되기 때문에, 동시에 접근하는 것을 방지하기 위해 lock을 걸어줄 것이다. 곧장 걸고, 전체 코드를 보자.

4. 완성

import socket, array
from threading import Lock, Thread

import matplotlib.pyplot as plt
import matplotlib.animation as animation

class RealTimeAnimation:
    """
    Animate original voltage signal graph and fft result.
    """
    def __init__(self):
        # define plot
        self.fig = plt.figure(figsize=(16, 6))
        self.original = self.fig.add_subplot(1, 1, 1)
        # connect animation method
        self.animation = animation.FuncAnimation(self.fig, self.update, interval=100)

    def update(self, i) -> None:
        """
        Run every interval and update graphs.
        Using lock.

        Args:
            i : interval
        """

        # update original graph
        data_lock.acquire()
        if len(entire_data) > 110000:
            graph_data = entire_data[len(entire_data) - 100000:]
        else: 
            graph_data = entire_data
        data_lock.release()

        self.original.clear()
        self.original.plot(graph_data)

def run() -> None:
    """
    UDP data receive function.
    Using lock.
    """

    print("Connected", flush=True)

    while True:
        data = serverSocket.recv(16384)
        signals = array.array('f')
        signals.frombytes(data)
        data_lock.acquire()
        entire_data.extend(signals)
        data_lock.release()


if __name__ == "__main__":
    
    # define socket setting and bind
    server = ('10.0.0.119', 2001)
    serverSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    serverSocket.bind(server) 

    # set data / data lock
    entire_data = [0]
    data_lock = Lock()

    # start UDP in another thread
    t = Thread(target=run, args=())
    t.start()
    
    # start animation 
    animation = RealTimeAnimation()
    plt.show()

y에 대해 auto scaling이 되고, 10만 개 씩 데이터를 plotting 해주는 udp 기반 프로그램이 완성되었다. 

참고로 keyboard interrupt 등을 통해서는 이 프로그램을 종료할 수 없다(창은 꺼진다) 왜냐하면 thread 뒤쪽에서 소켓 통신이 무한루프를 돌고 있기 때문이다(...) 직접 실행 중인 커널을 삭제해서 종료해주어야한다. 좀 더 유연한 기법으로 thread를 join 시키는 방법을... 찾아보도록 하겠다.

다음 포스팅에서는 실시간 fft, 그리고 정지/이어서 실행, 데이터 저장, 저장한 데이터를 기반으로 fft, stft결과를 보여주는 추가 프로그램을 작성해보도록 하겠다.