Moth 연동 예제

정보

본 가이드에서는 Moth 서버가 구축되어 있는 상태를 전제로, 연동을 위한 예제 코드와 코드에 대한 설명을 제공합니다. Moth 서버 구축 방법은 시작하기-서버 구축를 참고하세요.

더 복잡한 커스터마이징을 원하시면 아래 예제를 참고하여 직접 Agent와 웹 UI를 개발할 수 있습니다. 전체 예제 코드는 Github에 업로드 된 Example Repository에서 확인할 수 있습니다. GitHub 예제에는 Lidar 연동 예제가 함께 제공됩니다.

로봇 연동 예제

ROS 토픽을 구독하여 영상 프레임을 수신하고, GStreamer로 h.264 코덱으로 인코딩하여 Moth로 전달하는 예제입니다.

초기화

생성자

이번 예제에서 사용될 Video객체의 생성자입니다. 연동을 위해 다음 변수 값 설정이 필요합니다.

• host_ : Moth 서버의 주소
• port_ : Moth 서버의 포트 번호
• ROS TOPIC(/your/image_raw/topic) : 로봇에서 영상 프레임을 전송하는 ROS 토픽

Video::Video() : Node("video")
{
  host_ = "your_host";
  port_ = "your_port";
  text = "video/h264;width=640;height=480;framerate=30;codecs=avc1.42002A";

  while (!initWebSocket()){
    std::cout << "reconnecting..." << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(1));
  }

  initVideo();

  RCLCPP_INFO(this->get_logger(), "Node has been started.");
  subscription_ = this->create_subscription<sensor_msgs::msg::Image>("/your/image_raw/topic", 10, std::bind(&Video::video_callback, this, std::placeholders::_1));
}

Websocket 초기화

Moth 서버로 연결하기 위한 웹소켓 초기화 코드입니다. 다음 값을 설정해 주어야 합니다.

• "your_path" : Moth에 웹소켓으로 연결하기 위한 도메인 이하 Path값을 설정합니다. 자세한 내용은 Moth 서버의 API 문서를 참고하세요.

bool Video::initWebSocket(){
  try{
    tcp::resolver resolver{ioc}; // HOST, PORT의 값을 ip로 변환하기 위한 resolver 생성
    ws_ptr = std::make_shared<websocket::stream<tcp::socket>>(ioc); // 웹소켓 포인터 객체 생성

    auto const results = resolver.resolve(host_, port_); // ip 변환
    auto ep = net::connect(ws_ptr->next_layer(), results); // ip에 접근

    std::string host = host_ + ':' + std::to_string(ep.port());

    ws_ptr->set_option(websocket::stream_base::decorator( // HTTP 헤더 설정
      [](websocket::request_type& req)
      {
        req.set(http::field::user_agent,
        std::string(BOOST_BEAST_VERSION_STRING) +
        " websocket-client-coro");
      }));
    ws_ptr->handshake(host, "your_path"); // 웹소켓 PATH 설정

    ws_ptr->write(net::buffer(text)); // MIME 전송 (현재 String)
    ws_ptr->binary(true); // 이후 데이터 바이너리로 전송

    std::cout << "connect" << std::endl;
    return true;
  } catch (...){
    std::cout << "ERR initWebSocket()" << std::endl;
    return false;
  }
}

GStreamer 초기화

영상 인코딩을 위해 GStreamer를 초기화하는 코드입니다. appsrc파이프라인으로 프레임을 전달하면 GStreamer에서 인코딩합니다. 인코딩이 완료된 프레임은 appsink로 전달되고, appsink_callback()함수에서 프레임을 서버로 전송합니다.

void Video::initVideo(){
  // appsrc: 영상 데이터를 직접 넣어주는 source, videoconvert: 비디오 데이터를 원하는 형식으로 변환 (지금은 기본으로), openh264enc: H.264 코덱으로 이미지를 인코딩 하는 하드웨어 인코더, h264parse: NAL Unit 및 SPS/PPS 정보를 담기 위한 파서, appsink: 데이터를 직접 받을 수 있는 sink
  const char* pipeline_cmd = "appsrc name=src do-timestamp=true is-live=true emit-signals=true format=time caps=video/x-raw,format=RGB,width=640,height=480 ! videoconvert ! openh264enc bitrate=1000000 ! video/x-h264, profile=high, alignment=au, stream-format=byte-stream ! h264parse config-interval=1 ! queue leaky=2 ! appsink name=sink sync=false drop=true emit-signals=true max-buffers=3";
  pipeline = gst_parse_launch(pipeline_cmd, NULL); // 파이프라인을 pipeline_cmd기반으로 생성
  appsrc = gst_bin_get_by_name(GST_BIN(pipeline), "src"); // src라는 이름의 오브젝트를 가져와 appsrc에 저장
  appsink = gst_bin_get_by_name(GST_BIN(pipeline), "sink"); // sink라는 이름의 오브젝트를 가져와 appsink에 저장
  g_signal_connect(appsink, "new-sample", G_CALLBACK(appsink_callback), this); // appsink의 출력을 on_new_sample_static이라는 callback함수에 전달
  gst_element_set_state(pipeline, GST_STATE_PLAYING); // 파이프라인의 상태를 PLAYING으로 변환, 즉 파이프라인 시작
}

ROS 프레임 수신

ROS 토픽을 구독하여 프레임을 수신하는 콜백함수입니다. 수신된 프레임을 GStreamer appsrc 파이프라인에 전달합니다.

void Video::video_callback(const sensor_msgs::msg::Image::SharedPtr msg){
  GstBuffer* buffer = gst_buffer_new_allocate(NULL, msg->data.size(), NULL); // ROS를 통해 들어온 msg 데이터의 size 만큼 GstBuffer 객체 할당
  GstMapInfo map; // buffer의 데이터를 매핑시킬 GstMapInfo 객체 생성
  gst_buffer_map(buffer, &map, GST_MAP_WRITE); // buffer의 주소값을 map의 주소값에 매핑
  std::memcpy(map.data, msg->data.data(), msg->data.size()); / map에 ROS 메세지의 데이터를 size만큼 복사
  GstFlowReturn ret;
  g_signal_emit_by_name(appsrc, "push-buffer", buffer, &ret); // buffer의 위치에 저장된 map 데이터를 GStreamer appsrc 파이프라인에 전달
  gst_buffer_unmap(buffer, &map);
  gst_buffer_unref(buffer);
  if (ret != GST_FLOW_OK) {
    g_printerr("Failed to push buffer to appsrc.\n");
  }
}

GStreamer 프레임 인코딩

GStreamer에서 인코딩된 프레임을 Moth 서버로 전송하는 콜백함수입니다.

GstFlowReturn Video::appsink_callback(GstAppSink *appsink, gpointer user_data){
  Video *self = static_cast<Video *>(user_data);
  return self->send_frame(appsink);
}

GstFlowReturn Video::send_frame(GstAppSink *appsink){
  GstSample *sample = gst_app_sink_pull_sample(GST_APP_SINK(appsink)); // appsink에서 나오는 데이터를 sample 객체에 저장
  if (!sample) {
    std::cerr << "Failed to pull sample" << std::endl;
    return GST_FLOW_ERROR;
  }
  GstBuffer *buffer_ = gst_sample_get_buffer(sample); // sample에서 GstBuffer 객체를 뽑아 buffer_에 저장
  GstMapInfo map_; // buffer에서 꺼낼 GstMapInfo라는 매핑 객체 생성
  if (gst_buffer_map(buffer_, &map_, GST_MAP_READ)){ // buffer_의 데이터를 map_에 매핑
    try{
      if (ws_ptr && ws_ptr->is_open() && (ws_ptr->next_layer().available()==0)){
        ws_ptr->write(net::buffer(map_.data, map_.size)); // map 데이터를 웹소켓으로 MOTH 서버에 전송
      }
    } catch (const beast::system_error& se) {
      std::cerr << "Broken pipe error. reconnecting..." << std::endl;
      exit(0);
    } catch (...) {
      std::cout << "unknown Error" << std::endl;
    }
    gst_buffer_unmap(buffer_, &map_);
  }
  gst_sample_unref(sample);
  return GST_FLOW_OK;
}

웹 연동 예제

웹소켓을 통해 h.264로 인코딩된 비디오프레임을 수신하여 디코딩하고 화면에 표시하는 예제입니다.

script.js

웹소켓을 통해 Moth와 연결하고, 수신된 데이터를 디코딩하여 화면에 표시하는 스크립트입니다.

const videoElement = document.getElementById("videoElement");

const textEncoder = new TextEncoder();
let mime, mimeObject;
let videoStreamTrack, videoWriter;

const HOST = ""; //Moth Host 정보를 입력하세요
const PORT = ""; //Moth Port 정보를 입력하세요
const CHANNEL_ID = ""; //Moth Channel 정보를 입력하세요
const TRACK = ""; //Moth Track 정보를 입력하세요

// 페이지 로드 시 실행되는 메인 함수
window.onload = async () => {
    videoStreamTrack = new MediaStreamTrackGenerator({
        kind: "video",
    });
    videoWriter = videoStreamTrack.writable.getWriter();
    await videoWriter.ready;

    const mediaStream = new MediaStream([videoStreamTrack]);
    videoElement.srcObject = mediaStream;

    connectWebsocket();
};

// websocket으로 moth연결
const connectWebsocket = () => {
    const socket = new WebSocket(
        `wss://${HOST}:${PORT}/pang/ws/sub?channel=${CHANNEL_ID}&track=${TRACK}&mode=bundle`
    );
    socket.binaryType = "arraybuffer";

    socket.onopen = () => {
        console.log("Connected to server");
        // 연결 유지를 위해 10초마다 ping 전송
        setInterval(() => {
            socket.send(textEncoder.encode("ping"));
        }, 10000);
    };

    socket.onmessage = async (event) => {
        if (typeof event.data === "string") {
            // mime 수신 시 video decoder 설정
            if (mime === event.data) return; // 기존 MIME과 동일하면 처리 생략

            mime = event.data;
            mimeObject = mimeStringToMimeObject(mime);

            // decode된 video frame을 video stream에 write
            const handleVideoFrame = async (frame) => {
                console.log("handleVideoFrame");
                if (frame && videoStreamTrack) {
                    videoWriter.write(frame);
                    frame.close();
                }
            };

            await setDecoder(mimeObject, handleVideoFrame);
        } else if (typeof event.data === "object") {
            // mime 들어오기 전 data 무시 & ping 수신 시 무시
            if (!mimeObject || new TextDecoder().decode(event.data) === "ping") {
                return;
            }
            // video data 수신 시 video decode
            await startDecode(processVideoData(event, mimeObject));
        }
    };

    socket.onclose = () => {
        console.log("Disconnected from server");
    };

    socket.onerror = (error) => {
        alert(`Error: ${error.message}`);
    };
};

// mime string을 사용하고자하는 mime object로 변환
const mimeStringToMimeObject = (mimeString) => {
    const [mimeType, ...mimeOption] = mimeString.split(";");

    const mimeOptionObj = mimeOption.reduce((acc, option) => {
        const [key, value] = option.trim().split("=");
        acc[key] = value;
        return acc;
    }, {});

    if (!mimeOptionObj.codec) mimeOptionObj.codec = mimeOptionObj.codecs;

    // 데이터 모드 설정 (데이터 앞에 붙은 정보가 있는지)
    if (mimeType.includes("+")) {
        const mode = mimeType.split("/")[1].split("+")[0];
        mimeOptionObj.data_mode = mode;
    }
    const type = mimeType.split("/")[0];
    mimeOptionObj.data_type = type;

    return mimeOptionObj;
};

// video data를 decode하기 위해 데이터 정리
const processVideoData = (event, mime) => {
    let sliceNumber = 0;

    if (mime.data_mode === "seq") {
        sliceNumber = 1;
    } else if (mime.data_mode === "ts") {
        sliceNumber = 8;
    }

    const data = new Uint8Array(event.data.slice(sliceNumber));

    // 이미지(JPEG) 데이터의 경우 객체로 반환
    if (mime.data_type === "image") {
        return {
            data,
            timestamp: event.timeStamp,
        };
    }

    // 비디오 데이터의 경우 EncodedVideoChunk 객체로 변환
    const encodedChunk = new EncodedVideoChunk({
        type: data.length > 100000 ? "key" : "delta",
        data: data,
        timestamp: event.timeStamp,
        duration: 0,
    });

    return encodedChunk;
};

decoder.js

웹코덱을 이용해 코덱을 디코딩하는 보조 함수입니다.

let videoDecoder = null;
let decoderConfig = null;
let useWebCodec = true;
let handleVideoFrame = null;

// decoder 초기 설정
const setDecoder = async (config, handle) => {
    handleVideoFrame = handle;
    decoderConfig = config;

    // image codec을 사용하는지 확인
    if (decoderConfig.data_type === "image") {
        useWebCodec = false;
    } else {
        useWebCodec = true;

        // video codec을 사용하는 경우 decoder 설정이 유효한지 확인
        if (await isConfigSupported(decoderConfig).supported) {
            return;
        }

        videoDecoder = new VideoDecoder({
            // decode 후 나온 frame 처리
            output: async (frame) => {
                handleFrame(frame);
            },
            // decoder error 처리
            error: (error) => {
                console.error("Decoder error:", error);
                decoder.close();
            },
        });

        // decoder 설정
        videoDecoder.configure(decoderConfig);
        console.log("Decoder", videoDecoder);
    }
};

const isConfigSupported = async (config) => {
    return await VideoDecoder.isConfigSupported(config);
};

const handleFrame = async (frame) => {
    handleVideoFrame(frame);
    frame.close();
};

const startDecode = async (encodedVideoChunk) => {
    if (useWebCodec && typeof encodedVideoChunk === "object") {
        // video codec을 사용하는 경우 decoder로 decode
        try {
            if (!videoDecoder) return;

            if (videoDecoder.state === "closed") {
                await setDecoder(decoderConfig, handleVideoFrame);
            } else {
                videoDecoder.decode(encodedVideoChunk);
            }
        } catch (error) {
            // decoder error 발생 시
            console.error("Decoder error:", error);
            await setDecoder(decoderConfig, handleVideoFrame);
        }
    } else {
        // image codec을 사용하는 경우 blob을 imageBitmap으로 변환 후 frame 처리
        const videoChunk = encodedVideoChunk;
        const blob = new Blob([videoChunk.data], {type: "image/jpeg"});

        createImageBitmap(blob)
            .then((imageBitmap) => {
                const decodedChunk = new VideoFrame(imageBitmap, {
                    timestamp: videoChunk.timestamp,
                });
                handleVideoFrame(decodedChunk);
            })
            .catch((error) => {
                console.log("ImageBitmap creation error:", error);
            });
    }
};

index.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Video | CoBiz</title>
    <style>
        body {
            margin: 0;
            padding: 0;
            box-sizing: border-box;
        }

        video {
            width: 100%;
            height: 100%;
            background-color: black;
        }
    </style>
</head>
<body>
<video id="videoElement" autoplay muted controls></video>
<script src="scripts/decoder.js"></script>
<script src="scripts/script.js"></script>
</body>
</html>