gRPC Frame 開発ガイド / gRPC Quick Start / イベント配信 / デバイスサーバー開発
gRPC Frame は、通常のメソッド呼び出し Describe / Call に加え、複数の結果を順次受信する StreamCall を提供する。Call の対象は ctrl_obj、StreamCall の対象は stream_obj である。stream_obj は省略可能であり、ストリーミングを使う場合に指定する。
DeviceClass ─────────── ctrl_obj ── Describe / Call ── GrpcClient
│
└─ 測定結果 ── stream_obj ── StreamCall ── client.stream_call(...)
stream_obj の公開メソッドはイテレータを返す。gRPC Frame は各 yield の値を adapter.pack_result() で直列化し、クライアントは adapter.unpack_result() で復元する。通常の Describe から StreamCall 用メソッドが自動バインドされるわけではないため、クライアントから stream_call(メソッド名, ...) を明示的に呼ぶ。
MiniPIX の pixet_grpc_server.py では、制御オブジェクトとストリームオブジェクトを分離して登録している。
from grpc_frame.grpc_server import create_grpc_server
ctrl = MiniPixCtrl(logger=logger, log_level=log_level)
ctrl.connect()
stream_obj = _MiniPixStream(ctrl)
server = create_grpc_server(
ctrl_obj=ctrl,
stream_obj=stream_obj,
logger=logger,
log_level=log_level,
max_workers=10,
)
ここで _MiniPixStream は MiniPIX パッケージ側に実装したクラスであり、Frame の組み込みクラスではない。DeviceClass が提供する測定コールバックを queue.Queue に入れ、別スレッドで取得処理を動かしながら、キューから順次 yield する。
# _MiniPixStream の処理の要点（MiniPIX 実装から抜粋）
def acquire_frames(self, count: int, sec: float):
yield from self._acquire(count, sec, "frames")
def acquire_integral(self, count: int, sec: float):
yield from self._acquire(count, sec, "integral")
_acquire() は、測定フレームを受け取る frame_callback、取得を実行する worker、終了・例外を知らせるキュー項目を持つ。コールバックとgRPCの送信処理を同じスレッドに押し込まない構成である。
from grpc_frame.grpc_client import GrpcClient
client = GrpcClient(server_ip="127.0.0.1", server_port=50051)
try:
for frame in client.stream_call("acquire_frames", 3, 0.1):
# frame は stream_obj.acquire_frames() が yield した値
print(type(frame))
finally:
client.close()
接続先と測定条件は使用するデバイスサーバーに合わせる。stream_call() はジェネレータを返す。途中で受信を終了するときはイテレータの close() を利用でき、Frame の実装では終了処理でgRPC呼び出しの cancel() を行う。サーバー側もクライアント切断を context.is_active() で検知し、イテレータの close() を呼ぶ。
MiniPIX の専用クライアントは、取得結果の保存先をクライアント側にするために acquire_frames() / acquire_integral() をオーバーライドしている。AsyncPixetGrpcClient は動的に結び付いたリモートメソッドを _grpc_acquire_frames / _grpc_acquire_integral に退避し、公開メソッドをローカル保存処理付きに置き換える。同期版 SyncPixetGrpcClient も同じ方針で実装されている。
# AsyncPixetGrpcClient の実装から抜粋
result = await self._grpc_acquire_frames(
count, sec, save_format=save_format, save_path=save_path,
)
return self._finish_acquisition(
result, name, save_local, return_data,
)
この保存機能は 通常の Call で取得した結果をクライアント側に保存する実装であり、前節の StreamCall とは別の経路である。画像転送とストリーミングを同一視しない。MiniPIX の保存処理では TIFF は数値配列を保持し、PNG は表示用16-bitグレースケールに変換する。
大きな画像を送る場合、create_grpc_server(grpc_options=...) およびクライアント側の grpc_options により、gRPCメッセージサイズを調整できる。測定フレームのサイズ、受信側の処理速度、キューの蓄積、切断時の後始末を考慮する。
mini_pix/src/mini_pix/server_grpc/pixet_grpc_server.py：_MiniPixStream、stream_obj の登録。mini_pix/src/mini_pix/server_grpc/async_pixet_grpc_client.py：リモートメソッドのオーバーライドと画像保存。mini_pix/src/mini_pix/server_grpc/sync_pixet_grpc_client.py：同期クライアント。grpc_frame/grpc_server.py：StreamCall のディスパッチと送信。grpc_frame/grpc_client.py：stream_call() の受信とキャンセル。