image_server.server_grpc package

class ImageStreamClient(*args, **kwargs)

ベースクラス: SyncGrpcClient

ImageStreamGrpcService用同期クライアント。

static decode_payload(payload)

パラメータ:

payload (Dict[str, Any])

戻り値の型:

Any

iter_frames(topic='camera.frame')

パラメータ:

topic (str)

戻り値の型:

Iterator[Tuple[Dict[str, Any], Any]]

show_camera_frame_stream(topic='camera.frame', window_name='camera.frame')

パラメータ:

• topic (str)

• window_name (str)

戻り値の型:

None

class ImageStreamGrpcService(image_stream=None, topic_name='camera.frame', publish_interval_sec=0.1, jpeg_quality=85, logger=None)

ベースクラス: object

ローカル画像制御をgRPC公開およびEventBus配信へ接続する。

_encode_frame(frame)

パラメータ:

frame (Any)

戻り値の型:

Dict[str, Any]

_publish_loop()

戻り値の型:

None

_set_event_bus(event_bus)

grpc_frameからEventBusを設定する。

パラメータ:

event_bus (Any)

戻り値の型:

None

static _time_ns()

戻り値の型:

int

static _validate_jpeg_quality(value)

パラメータ:

value (int)

戻り値の型:

int

capture_once()

最新画像をJPEGペイロードとして返す。

戻り値の型:

Dict[str, Any]

clear_error()

戻り値の型:

bool

connect()

ローカル画像取得デバイスへ接続する。

戻り値の型:

bool

disconnect()

配信を停止し、ローカル画像取得デバイスを切断する。

戻り値の型:

bool

get_config()

ローカル設定と配信設定を返す。

戻り値の型:

Dict[str, Any]

get_status()

ローカル取得状態とgRPC配信状態を統合して返す。

戻り値の型:

Dict[str, Any]

is_connected()

戻り値の型:

bool

publish_once()

最新画像をEventBusへ1回配信する。

戻り値の型:

bool

save_snapshot(file_path)

サーバー側ファイルへ最新画像を保存する。

パラメータ:

file_path (str)

戻り値の型:

Dict[str, Any]

set_camera_config(width=None, height=None, fps=None)

パラメータ:

• width (int | None)

• height (int | None)

• fps (float | None)

戻り値の型:

Dict[str, Any]

set_camera_id(camera_id)

パラメータ:

camera_id (int)

戻り値の型:

bool

set_stream_config(interval_sec=None, jpeg_quality=None, topic_name=None)

gRPC配信設定を変更する。

パラメータ:

• interval_sec (float | None)

• jpeg_quality (int | None)

• topic_name (str | None)

戻り値の型:

Dict[str, Any]

start_stream()

EventBusへのJPEG画像配信を開始する。

戻り値の型:

bool

stop_stream()

EventBusへの画像配信を停止する。

戻り値の型:

bool

Submodules

image_server.server_grpc.client module

grpc_frame用同期画像クライアント。

class ImageStreamClient(*args, **kwargs)

ベースクラス: SyncGrpcClient

ImageStreamGrpcService用同期クライアント。

static decode_payload(payload)

パラメータ:

payload (Dict[str, Any])

戻り値の型:

Any

iter_frames(topic='camera.frame')

パラメータ:

topic (str)

戻り値の型:

Iterator[Tuple[Dict[str, Any], Any]]

show_camera_frame_stream(topic='camera.frame', window_name='camera.frame')

パラメータ:

• topic (str)

• window_name (str)

戻り値の型:

None

image_server.server_grpc.image_stream_grpc_service module

ImageStreamDeviceClassをgrpc_frameへ公開するサービス層。

class ImageStreamGrpcService(image_stream=None, topic_name='camera.frame', publish_interval_sec=0.1, jpeg_quality=85, logger=None)

ベースクラス: object

ローカル画像制御をgRPC公開およびEventBus配信へ接続する。

_encode_frame(frame)

パラメータ:

frame (Any)

戻り値の型:

Dict[str, Any]

_publish_loop()

戻り値の型:

None

_set_event_bus(event_bus)

grpc_frameからEventBusを設定する。

パラメータ:

event_bus (Any)

戻り値の型:

None

static _time_ns()

戻り値の型:

int

static _validate_jpeg_quality(value)

パラメータ:

value (int)

戻り値の型:

int

capture_once()

最新画像をJPEGペイロードとして返す。

戻り値の型:

Dict[str, Any]

clear_error()

戻り値の型:

bool

connect()

ローカル画像取得デバイスへ接続する。

戻り値の型:

bool

disconnect()

配信を停止し、ローカル画像取得デバイスを切断する。

戻り値の型:

bool

get_config()

ローカル設定と配信設定を返す。

戻り値の型:

Dict[str, Any]

get_status()

ローカル取得状態とgRPC配信状態を統合して返す。

戻り値の型:

Dict[str, Any]

is_connected()

戻り値の型:

bool

publish_once()

最新画像をEventBusへ1回配信する。

戻り値の型:

bool

save_snapshot(file_path)

サーバー側ファイルへ最新画像を保存する。

パラメータ:

file_path (str)

戻り値の型:

Dict[str, Any]

set_camera_config(width=None, height=None, fps=None)

パラメータ:

• width (int | None)

• height (int | None)

• fps (float | None)

戻り値の型:

Dict[str, Any]

set_camera_id(camera_id)

パラメータ:

camera_id (int)

戻り値の型:

bool

set_stream_config(interval_sec=None, jpeg_quality=None, topic_name=None)

gRPC配信設定を変更する。

パラメータ:

• interval_sec (float | None)

• jpeg_quality (int | None)

• topic_name (str | None)

戻り値の型:

Dict[str, Any]

start_stream()

EventBusへのJPEG画像配信を開始する。

戻り値の型:

bool

stop_stream()

EventBusへの画像配信を停止する。

戻り値の型:

bool

image_server.server_grpc.make_pyi module

ImageStreamClientの型スタブを生成する。

main()

grpc_frameのgenerate_client_pyi()でclient.pyiを生成する。