iT邦幫忙

2025 iThome 鐵人賽

DAY 3
0
自我挑戰組

Robot Framework 與 Websocket 協議測試系列 第 9

Protobuf Python 端的編譯與序列化/反序列化/編解碼處理

  • 分享至 

  • xImage
  •  

前言

抱歉,因為最近工作繁忙,斷更了一週,現在就繼續為大家補上精彩內容。

上篇聊到 Protobuf 的命令訊息結構,之前也有講過它有豐富的 SDK 可以套用。
那今天就以 Python 為例解釋一下我們怎麼在 Python 端做序列化跟反序列化的處理。

序列化與反序列化

稍微提一下這個概念,所謂的序列化跟反序列化,就是本地的系統處理歸本地系統處理,透過傳輸管道、頻道或者通道傳輸時,就需要做一個序列化的動作,讓它變成一個可以串流輸出的資料流。
反過來,反序列化就是把串流傳出去的資料再組合成程式端可以理解、有邏輯架構的資料結構。

步驟一: 編譯 .proto 訊息介面定義檔為 Python 用

第一步就是將服務端定義好的 .proto 檔案編譯成 Python 可以引用的模組檔 .py 檔案。
除此之外,如果想要給如 Pycharm 的 IDE 引用檢查的時候,還可以額外生成.pyi,Python 的類別提示檔案。

第一步: 到 Github 官網 ,下載電腦端的套件,以下會以 Windows 為例

第二步: 執行 protoc (Protobuf的編譯器)
protoc --python_out=. --pyi_out=. proto/game_messages.proto
簡單拆解一下:
protoc: Protobuf的編譯器
--python_out=.: 要求它生成 Python 可用的命令訊息模組 (.py)
--pyi_out=.: 要求它生成 Python 的類別提示檔案 (.pyi)
proto/game_messages.proto: 最後是要編譯的 proto 檔案

PS C:\Users\Scott\PycharmProjects\sample_gaming_sut> dir .\proto\*

    Directory: C:\Users\Scott\PycharmProjects\sample_gaming_sut\proto

Mode                 LastWriteTime         Length Name
----                 -------------         ------ ----
d----         2025/8/16 下午 06:03                __pycache__
-a---         2025/8/16 下午 06:02           2166 game_messages.proto

PS C:\Users\Scott\PycharmProjects\sample_gaming_sut> protoc --python_out=. --pyi_out=. proto/game_messages.proto

PS C:\Users\Scott\PycharmProjects\sample_gaming_sut> dir .\proto\*

    Directory: C:\Users\Scott\PycharmProjects\sample_gaming_sut\proto

Mode                 LastWriteTime         Length Name
----                 -------------         ------ ----
d----         2025/8/16 下午 06:03                __pycache__
-a---         2025/9/28 下午 03:40           5488 game_messages_pb2.py
-a---         2025/9/28 下午 03:40           7806 game_messages_pb2.pyi
-a---         2025/8/16 下午 06:02           2166 game_messages.proto

打開編譯好的 py 看一眼
_descriptor_pool 是一堆二進位描述的命令訊息規格
下面的 _globals['_XXX']._serialized_start=1333 & _globals['_XXX']._serialized_end=1411 就是告訴 Protobuf 要去哪裡看 XXX 命令對應的輸入跟輸出規格,_descriptor_pool 看不懂沒關係,知道他怎麼找資料就好

# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: proto/game_messages.proto
# Protobuf Python Version: 6.32.0
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
    _runtime_version.Domain.PUBLIC,
    6,
    32,
    0,
    '',
    'proto/game_messages.proto'
)
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()




DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x19proto/game_messages.proto.....proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'proto.game_messages_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
  DESCRIPTOR._loaded_options = None
  _globals['_GAMEROUNDSTATUS']._serialized_start=1333
  _globals['_GAMEROUNDSTATUS']._serialized_end=1411
  _globals['_LOGINREQUEST']._serialized_start=37
  _globals['_LOGINREQUEST']._serialized_end=87
  ...
  _globals['_ERRORRESPONSE']._serialized_start=1256
  _globals['_ERRORRESPONSE']._serialized_end=1331
# @@protoc_insertion_point(module_scope)

步驟二: Python 中如何透過 Protobuf 處理資料序列化跟反序列化

先從 import 引用看起
這裡的 game_messages_pb2 就是對應上一步驟生成的 "game_messages_pb2.py" 檔案
後續就用 pb 來跟 Protobuf 訊息互動

try:
    from proto import game_messages_pb2 as pb
except ImportError:
    print("Error: Protocol buffer files not generated. Run: protoc --python_out=. --pyi_out=. proto/game_messages.proto")
    sys.exit(1)

以下以登入命令的發送為例,首先複習一下 login 的命令格式:

// Authentication Messages
message LoginRequest {
    string username = 1;
    string password = 2;
}

message LoginResponse {
    bool success = 1;
    string message = 2;
    string session_token = 3;
    int64 user_id = 4;
    int64 balance = 5;
}

要發送登入,步驟如下:

  1. 使用 pb.LoginRequest() 建立名為 request 的 LoginRequest 物件
  2. 需要使用者名稱跟帳戶兩個參數,把以上兩個參數指定到 request 對應的屬性中
  3. 呼叫 websocket 的發送指令
        request = pb.LoginRequest()
        request.username = username
        request.password = password

        await self.send_message(C2S_LOGIN_REQ, request)

要接收登入的回覆就反過來:

  1. 等待訊息接入
  2. 生成名為 response 的 LoginResponse 的物件
  3. 使用 ParseFromString 將序列化成十六進位的資料流,反序列化成 LoginResponse 需要的資料結構存到 response 中
  4. 檢查狀態碼是否正常?
  5. 狀態碼正常,再把 response 內回傳的資料一一解讀
        response_data = await self.receive_message()

        if response_data['command_id'] == S2C_LOGIN_RSP:
            response = pb.LoginResponse()
            response.ParseFromString(response_data['payload'])

            if response.success:
                self.authenticated = True
                self.user_id = response.user_id
                self.balance = response.balance
                self.session_token = response.session_token
                logger.info(f"Login successful! User ID: {self.user_id}, Balance: {self.balance}")

小結

今天講的主題也很純粹,就是講如何編譯 .proto 與如何在 Python 中做序列化跟反序列化,並使用 login 這個登入處理作為範例。

下一章節就會講簡單的服務器與客戶端聯調,作爲一個基準線。


上一篇
待測系統命令訊息解析
下一篇
簡易的伺服器跟客戶端聯調
系列文
Robot Framework 與 Websocket 協議測試10
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言