[한이음_ROS2] 수집한 센서 데이터를 CSV 파일로 추출하기

개발환경

SBC : Raspberry Pi 4B 4GB

PC : Ubuntu20.04 LTS

설명

온습도 센서, 미세먼지 센서, 유해가스 센서로부터 수집한 데이터를 서브스크라이브하여 CSV 파일로 추출한다.

CSV파일의 데이터를 이용하여 인공지능 모델을 학습한다.

• 구분자는 ' , ' [컴마] 이다.
• 년 | 월 | 일 | 시간 | 구역 | 온도 | 습도 | 오염도

과정

소스코드

import rclpy
import pandas as pd 
import datetime as dt 
from rclpy.node import Node
from rclpy.qos import QoSDurabilityPolicy
from rclpy.qos import QoSHistoryPolicy
from rclpy.qos import QoSProfile
from rclpy.qos import QoSReliabilityPolicy
from rclpy.callback_groups import ReentrantCallbackGroup
from std_msgs.msg import Int16
from std_msgs.msg import Float32
from hanium_interface.msg import GetDht

class DataExport(Node):
    
    def __init__(self):
        super().__init__("DataExport")
        
        self.df = pd.DataFrame(columns=['year','month','day', 'hour','local' , 'temp', 'humidity', 'metter'])
        time_now = dt.datetime.now()
        
        self.year = str(time_now.year)
        self.month = str(time_now.month)
        self.day = str(time_now.day)
        self.hour = str(time_now.hour)
        self.temp = 0.0
        self.hum = 0.0
        self.pollu = 0
        
        QOS_RKL10V = QoSProfile(
            reliability=QoSReliabilityPolicy.RELIABLE,
            history=QoSHistoryPolicy.KEEP_LAST,
            depth=10,
            durability=QoSDurabilityPolicy.VOLATILE)

        self.subscribe_hum = self.create_subscription(
             GetDht,
             'dht_sensor',
             self.get_hum_value,
             QOS_RKL10V
             )
             
        self.subscribe_pollu = self.create_subscription(
            Int16,
            'pollu_grade',
            self.get_pollu_value,
            QOS_RKL10V
        )
        
        self.subscribe_space = self.create_subscription(
            Int16,
            'space_num',
            self.get_space_value,
            QOS_RKL10V
        )
        
        
    
    def get_hum_value(self, msg):
        
        self.hum = round(msg.hum,1)
        self.temp = round(msg.temp,1)
        self.get_logger().info('습도값은 : {0}, 온도값은 : {1}'.format(self.hum,self.temp))
        
    def get_pollu_value(self, msg):
    
        self.pollu = msg.data
        self.get_logger().info('오염도 수치는 : {0}'.format(self.pollu))
        
    def get_space_value(self, msg):
    
        self.space = msg.data
        self.df.loc[self.space] = [self.year,self.month,self.day,self.hour, self.space, self.temp, self.hum, self.pollu]
        self.df.to_csv('DataExport'+'_'+ self.year + self.month + self.day +'.csv',index = False,sep=',')
        self.get_logger().info('구역 번호는 : {0}'.format(self.space))
        
def main(args=None):
    rclpy.init(args=args)
    
    try:
        export = DataExport()
        try:
            rclpy.spin(export)
        except KeyboardInterrupt:
            export.get_logger().info('Keyboard Interrupt (SIGINT)')
        finally:
            export.destroy_node()
    finally:
        rclpy.shutdown()


if __name__ == '__main__':
    main()

소스코드 부분 분석

• 데이터 프레임을 위해 pandas를 import 하였다.
• 시간 정보를 얻기 위해서 datetime을 import 한다.
• msg는 Int16, Float32와 사용자 정의 메세지인 GetDht를 사용한다.

• 노드 이름을 "DataExport" 로 지정한다.
• pandas를 이용하여 데이터의 열 이름을 지정한다.
• 년, 월, 일, 시간, 온도, 습도, 오염도 데이터를 각각 초기화 해준다.

• QoS 파라미터를 지정해준다.
• 습도, 오염도, 구역번호 서브스크라이버를 선언해준다.

• 각 반복함수는 정해진 메세지를 서브스크라이브 한다.
• 온습도와 오염도등 데이터는 각 구역을 지날 때 마다 CSV파일에 순서대로 기록된다.

• 함수의 메인 구문이다.

결과

페이크 데이터 퍼블리시 결과

추출된 CSV파일 형태