gRPC Frame 開発ガイド / gRPC Quick Start / StreamCall と画像転送 / デバイスサーバー開発
gRPC Frame は EventBus によるサーバープロセス内のイベント配信と、gRPC Events/Subscribe によるクライアントへの通知を提供する。StreamCall は要求したメソッドが返す連続データを受信する仕組みであり、Subscribe はトピックに発行された通知を購読する仕組みである。
サーバープロセス クライアント
イベント発生元 → EventBus.publish() → Events/Subscribe → GrpcClient.subscribe()
│
└─ topic ごとの購読者へ配信
create_grpc_server() は event_bus を受け取り、省略した場合は内部で EventBus を生成する。生成したバスは Events サービスへ渡される。制御オブジェクトが _set_event_bus を持つ場合、Frame のサーバー実装はそのメソッドへバスを渡す。
EventBus は publish(topic, payload_obj, source="") を提供する。例えば通信から独立した機器制御クラスがイベント発生を通知する仕組みを持つ場合、サーバー側の連携コードでイベントを受け取り、バスへ発行できる。
from grpc_frame.event_bus import EventBus
from grpc_frame.grpc_server import create_grpc_server
bus = EventBus()
server = create_grpc_server(ctrl_obj=ctrl, event_bus=bus)
# サーバー側で条件成立を検知したときに発行する
bus.publish("measurement_ready", {"value": 12.3}, source="sensor")
上記の publish() はイベント発行APIの使用例である。実際のデバイスサーバーでは、測定スレッドなどのイベント発生元から呼び出す。DeviceClassの公開APIにgRPCの型や購読処理を追加する必要はない。
EventBus.subscribe() はサーバー内部のジェネレータであり、gRPCクライアントが直接呼ぶものではない。内部では購読者ごとにキューを持ち、Condition で通知を待つ。購読開始前に発行されたイベントを保持する履歴機構ではない。
from grpc_frame.grpc_client import GrpcClient
client = GrpcClient(server_ip="127.0.0.1", server_port=50051)
try:
for payload in client.subscribe(topic="measurement_ready", once=True):
print(payload)
finally:
client.close()
subscribe(topic=..., once=False, source="") は、受信したイベントの payload を順次返す。once=True はクライアントが1件を受信した時点で購読を終了する指定であり、測定装置側のイベント生成を停止する指定ではない。source 引数はリクエストに含まれるが、提供された EventBus の実装では発生元によるフィルタリングには使用されていない。
aandd_reader は、DeviceClassが持つ計測・閾値判定の機能と、gRPC Frameのイベント購読を組み合わせている。専用クライアント AanddReaderGrpcClient は、複数の機器操作を prepare_event_measurement() にまとめる。
# aandd_reader/server_grpc/aandd_reader_grpc_client.py の処理
self.stream_start()
self.event_set_config(threshold_g=threshold_g, above=above, topic=topic)
self.set_trigger_mode(
mode=trigger_mode,
window=trigger_window,
min_samples=trigger_min_samples,
alpha=trigger_alpha,
stable_only=trigger_stable_only,
)
self.event_arm(once=once)
stream_start() は aandd_reader側の計測処理を開始するメソッドであり、gRPCの StreamCall を開始するという意味ではない。event_set_config()、set_trigger_mode()、event_arm() も機器固有の公開メソッドである。once はここでは装置側のイベントarmの動作を指定する。
実際の利用例は次のとおり。
from aandd_reader import Config
from aandd_reader.server_grpc import AanddReaderGrpcClient
Config.load_yaml("config_server1.yaml")
client = AanddReaderGrpcClient(config=Config)
try:
client.prepare_event_measurement(
topic="weight_threshold",
threshold_g=10.0,
above=True,
once=True,
trigger_mode="raw",
)
event = client.wait_event()
print(event)
finally:
client.stream_stop()
client.close()
wait_event() は内部で subscribe(topic=設定したトピック, once=True) を実行して最初のイベントを返す。装置側の event_arm(once=...) と、通信側の subscribe(once=...) は別の設定である。trigger_mode には raw、mean、median、ema があり、平均・中央値の窓幅やEMA係数も専用クライアントから指定できる。
イベントの発生前に購読が成立している必要がある場合、準備・購読開始の順序を装置の動作に合わせて設計する。上記は提供された grpc_client_stream_event.py と専用クライアントの使用形態に対応する。
grpc_frame/event_bus.py：EventBus.publish() / subscribe()。grpc_frame/grpc_server.py：Events/Subscribe のサーバー実装。grpc_frame/grpc_client.py：GrpcClient.subscribe()。aandd_reader/src/aandd_reader/server_grpc/aandd_reader_grpc_client.py：イベント測定の準備と待機。aandd_reader/examples/grpc_client_stream_event.py：閾値イベントの受信例。