StreamCall / 画像転送 / EventBus / イベント購読
gRPC Frameによる開発手順を、DeviceClass、サーバー、クライアント、公開・型情報の順に説明する。MiniPIXとaandd_readerはFrameの使い方を示す実装例であり、装置別マニュアルではない。導入についてはインストールマニュアルを参照する。gRPCではApiSpecを用いず、自動ディスパッチを使用する。
mini_pix/
├── pixet_ctrl.py # 機器制御本体
└── server_grpc/
├── __init__.py # DeviceProxy の公開窓口（別途実装）
├── __init__.pyi # DeviceProxy の型宣言
├── pixet_grpc_server.py # サーバー構築・起動
├── sync_pixet_grpc_client.py # 同期クライアント
├── async_pixet_grpc_client.py# 非同期クライアント
└── make_pyi.py # クライアント型宣言生成
この構成は主要なファイルの抜粋である。COBOTTA の server_fastapi/models/routers/clients/ と同じディレクトリ構造に揃える必要はない。
MiniPixCtrl は pixet_ctrl.py で定義され、gRPC、GUI、サーバー設定、クライアント設定を扱わない。connect() / disconnect()、acquire_frames() / acquire_integral()、測定結果の取得・保存を提供する。コンストラクタには logger が必要である。
from mini_pix.pixet_ctrl import MiniPixCtrl
ctrl = MiniPixCtrl(logger=logger)
ctrl.connect()
try:
# DeviceClass の操作は通信なしでも利用できる。
ctrl.acquire_frames(1, 1.0)
finally:
ctrl.disconnect()
logger は事前に用意する。上記は呼び出し関係を示す抜粋であり、PIXet の導入、装置接続、測定条件設定までを含む単独実行スクリプトではない。pixet_ctrl.py では PIXet の Windows 用 API ディレクトリが指定されているため、利用環境に応じた設定が必要である。
pixet_grpc_server.py は MiniPixCtrl を生成し、grpc_frame.grpc_server.create_grpc_server に渡す。サーバー用の継承クラスを必須とせず、DeviceClass をそのまま公開できる。
from grpc_frame.grpc_server import create_grpc_server
from mini_pix.pixet_ctrl import MiniPixCtrl
ctrl = MiniPixCtrl(logger=logger, log_level=log_level)
ctrl.connect()
server = create_grpc_server(
ctrl_obj=ctrl,
stream_obj=stream_obj,
logger=logger,
log_level=log_level,
max_workers=10,
)
server.add_insecure_port(f"{grpc_host}:{grpc_port}")
server.start()
これは元のサーバーの構成部分の抜粋である。stream_obj は次節の _MiniPixStream(ctrl) で生成し、終了時には server.stop(...) と ctrl.disconnect() を行う。実際の pixet_grpc_server() にはポートのバインド確認、起動後の接続確認、終了処理が含まれる。上記抜粋だけを起動用の完成スクリプトとして扱わない。
測定中のフレームを順次返すため、_MiniPixStream が MiniPixCtrl を保持し、acquire_frames() / acquire_integral() をジェネレータとして提供する。内部ではワーカースレッドとキューを用いて、DeviceClass の callback から受け取ったフレームを StreamCall に渡す。
stream_obj = _MiniPixStream(ctrl)
server = create_grpc_server(ctrl_obj=ctrl, stream_obj=stream_obj, ...)
... は説明用の省略記号であり、実際の呼び出しでは上記の logger などの引数を指定する。通常のメソッド呼び出しとストリームの提供は区別する。ストリームのために DeviceClass に gRPC の型を持ち込まない。
AsyncPixetGrpcClient(AsyncGrpcClient) は、サーバーから返る測定結果をクライアント側で TIFF / PNG として保存できるよう、acquire_frames() と acquire_integral() の動作を上書きする。SyncPixetGrpcClient も同期呼び出しに対して同様の処理を提供する。
# async_pixet_grpc_client.py の動作を示す抜粋
async def _acquire_frames_async(
self, count, sec, name="frame.tiff", *,
save_local=True, return_data=False,
save_format=None, save_path=None,
):
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)
クライアントの公開操作は acquire_frames(...) のままである。_grpc_acquire_frames はクライアント実装内で保持する元のリモート呼び出しへの参照であり、利用者が直接呼び出す公開 API ではない。save_local=True なら結果をクライアントのファイルシステムへ保存する。return_data=True なら結果全体を返し、そうでなければ result["rc"] を返す。TIFF は生の数値配列、PNG は表示用の16-bitグレースケールに変換する。複数フレームでは連番ファイル名を使う。
aandd_reader は、mini_pix のフレーム転送とは異なり、通常の測定メソッドとイベント購読を示す。以下はgRPC Frameの利用方法を説明する例であり、重量計の操作マニュアルではない。
ctrl_obj にDeviceClassインスタンスを渡すsrc/aandd_reader/server_grpc/aandd_reader_grpc_server.py の主要部分：
ctrl = AanddReader(port=device_port, logger=logger, log_level=log_level)
server = create_grpc_server(
ctrl_obj=ctrl,
logger=logger,
log_level=log_level,
max_workers=10,
)
bind_addr = f"{grpc_host}:{grpc_port}"
ret = server.add_insecure_port(bind_addr)
if ret == 0:
raise RuntimeError(f"bind failed: {bind_addr}")
server.start()
実ファイルでは、起動後の接続確認、待機ループ、server.stop(grace=None) と ctrl.close() も行う。ここでは stream_obj を渡していない。ストリーム用オブジェクトは常に必須ではない。
from grpc_frame.grpc_client import GrpcClient, AsyncGrpcClient
class AanddReaderGrpcClient(GrpcClient):
# 実ファイルでは接続設定・イベント用状態・操作メソッドを定義
...
class AsyncAanddReaderGrpcClient(AsyncGrpcClient):
# 実ファイルでは非同期ラッパーとイベント用操作を定義
...
このコードブロックは継承関係を示す説明用の省略であり、完全なクラス実装ではない。実装では、同期クライアントの prepare_event_measurement() が stream_start()、event_set_config()、set_trigger_mode()、event_arm() を順に呼び、wait_event() が購読結果を返す。非同期クライアントは await を用いて同じ準備処理を行い、購読待機は _run_in_thread() で実行する。
examples/grpc_client_stream_event.py では、クライアントが prepare_event_measurement(topic="weight_threshold", threshold_g=10.0, above=True, once=True, trigger_mode="raw") を呼び、続いて wait_event() で1件待機し、最後に stream_stop() を呼ぶ。個々のリモート操作を組み合わせた用途別の公開メソッドをクライアント継承クラスに置く例として参照できる。
examples/grpc_client_meas_async.py は AsyncAanddReaderGrpcClient(config=Config, logger=logger, log_level=log_level) を作り、await client.meas() を実行する。server_grpc/make_pyi.py とクライアントの .pyi は型情報のために配置されている。
ese774 Frameの例では FastApiServer(device_cls=..., router_cls=..., api_spec=..., device_kwargs=...)、gRPC Frameの例では create_grpc_server(ctrl_obj=...) を使用する。一方の起動引数を他方へそのまま移植しない。 両者の違いはDeviceClassに持ち込まず、通信層の実装として扱う。
measure_simple.py の主要部分は次のとおり。
from mini_pix.config import Config
from mini_pix.server_grpc import DeviceProxy
config = Config("mini_pix.yaml")
client = DeviceProxy(
"mini_pix",
server_ip=config.SERVER_IP,
server_port=config.SERVER_PORT,
set_timeout_sec=config.TIMEOUT,
async_mode=False,
)
client.connect()
try:
client.set_mode_all("Counting")
client.set_threshold(5.0)
client.set_bias(150)
client.set_clock(9.66667)
client.acquire_frames(
1, 1.0, name="measurement_output/frame.tiff",
save_local=True, return_data=False,
)
finally:
client.disconnect()
これは設定値と出力名を抜粋した説明例である。実際の measure_simple.py は時刻付きファイル名を作成し、XLogger を使用する。測定条件は装置に適合する値を確認して設定する。__init__.pyi には DeviceProxy("mini_pix", ..., async_mode=False/True) のオーバーロードがあり、同期・非同期の戻り値型をそれぞれ SyncPixetGrpcClient / AsyncPixetGrpcClient として宣言している。
server_grpc/make_pyi.py は grpc_frame.make_pyi.generate_client_pyi を利用して同期・非同期クライアントの .pyi を生成し、クライアント側で上書きしたメソッドの宣言を補正する。__init__.py と __init__.pyi は手書きの公開 API として保持し、生成・上書きしない。これはこのパッケージの実例であり、すべての gRPC DeviceClass に同じ生成スクリプトを要求するものではない。
.pyiをパッケージ側に用意する。内部メソッドの命名は公開I/Fの仕様ではない。このページは gRPC Frameを使ったデバイスサーバーの作り方 を説明する。mini_pix のソースコードは、構成と実装上の約束事を具体的に示すための参照例であり、特定の測定器の操作マニュアルではない。gRPC Frame は自動ディスパッチのみで、ese774 Frame の Pydantic / ApiSpec による明示API方式は使わない。