Zenn
📶

任意のバイトデータをGStreamerで受け取る(UDP)

に公開
Raspberry Pi
UDP
GStreamer
tech

サーバー側　Raspberry Pi Zero 2 W

Pythonコードです。

import sys
import gi
import struct

gi.require_version('GLib', '2.0')
gi.require_version('Gst', '1.0')

from gi.repository import Gst, GObject, GLib

pipeline = None
bus = None
message = None

# initialize GStreamer
Gst.init(sys.argv[1:])
loop = GLib.MainLoop()

def cb_udp(sink):
    sample = sink.emit("pull-sample")
    if not sample:
        return Gst.FlowReturn.ERROR

    buf = sample.get_buffer()
    pts = buf.pts
    result, map_info = buf.map(Gst.MapFlags.READ)
    if result:
        cnt = struct.unpack("<i", map_info.data) # 符号付き整数4byte,リトルエンディアン
        print("receive: ", cnt)
        buf.unmap(map_info)
        return Gst.FlowReturn.OK

def cb_message(bus, message): # 起動中、busからのメッセージ処理
    global loop
    msg_type = message.type

    # ストリーム終了時
    if msg_type == Gst.MessageType.EOS:
        print("EOS")
        loop.quit()
        
    # エラー時
    elif msg_type == Gst.MessageType.ERROR: 
        print("ERROR")
        loop.quit()

# build the pipeline　UDPでバイナリ受信
pipeline = Gst.parse_launch(
    "udpsrc port=9001 caps=\"application/x-raw,format=byte\" ! appsink name=sink"
)

# start playing
print("Starting pipeline...")
ret = pipeline.set_state(Gst.State.PLAYING)

# appsinkの設定
sink = pipeline.get_by_name("sink")
sink.set_property("emit-signals", True)
sink.connect("new-sample", cb_udp) # "new-sample"シグナルを受信するたびにコールバック

# busの方での終了処理監視
bus = pipeline.get_bus()
bus.add_signal_watch() 
bus.connect("message", cb_message)

try:
    print("Waiting Data...")
    loop.run() # 受信ループ
        
except KeyboardInterrupt:
    print("KeyboardInterrupt")
    loop.quit()
    pass

# free resources
print("Freeing pipeline...")
pipeline.set_state(Gst.State.NULL)
print("Finish")

クライアント側　Windows11

こちらの方のクライアント側のコードを参考に以下の変更を加えます。
https://zenn.dev/jinchangz/articles/e173d59c0f4dd0

import struct
import time

# ラズパイのアドレス
server_address = "192.168.137.183"

message = struct.pack("<i", 1) # b'\x01\x00\x00\x00' 受信側とサイズ合わせる

try:
  for i in range(10):
    message = struct.pack("<i", i)
    print('sending {!r}'.format(message))
    sent = sock.sendto(message, (server_address, server_port))
    print('Send {} bytes'.format(sent))
    time.sleep(1)
finally:
  print('closing socket')
  sock.close()

1秒間隔で0~9の値をUDP送信しています。

通信

サーバーを先に起動し、クライアントを実行すると、以下のように受信できます。

Starting pipeline...
Waiting Data...
receive:  (0,)
receive:  (1,)
receive:  (2,)
receive:  (3,)
receive:  (4,)
receive:  (5,)
receive:  (6,)
receive:  (7,)
receive:  (8,)
receive:  (9,)

Discussion