image_server.server_grpc package

class ImageStreamClient(*args, **kwargs)[source]

Bases: SyncGrpcClient

ImageStreamGrpcService用同期クライアント。

static decode_payload(payload)[source]

Parameters:: payload (Dict[str, Any])
Return type:: Any

iter_frames(topic='camera.frame')[source]

Parameters:: topic (str)
Return type:: Iterator[Tuple[Dict[str, Any], Any]]

show_camera_frame_stream(topic='camera.frame', window_name='camera.frame')[source]

Parameters:

topic (str)
window_name (str)

Return type:

None

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

Bases: object

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

_encode_frame(frame)[source]

Parameters:: frame (Any)
Return type:: Dict[str, Any]

_publish_loop()[source]

Return type:: None

_set_event_bus(event_bus)[source]

grpc_frameからEventBusを設定する。

Parameters:: event_bus (Any)
Return type:: None

static _time_ns()[source]

Return type:: int

static _validate_jpeg_quality(value)[source]

Parameters:: value (int)
Return type:: int

capture_once()[source]

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

Return type:: Dict[str, Any]

clear_error()[source]

Return type:: bool

connect()[source]

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

Return type:: bool

disconnect()[source]

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

Return type:: bool

get_config()[source]

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

Return type:: Dict[str, Any]

get_status()[source]

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

Return type:: Dict[str, Any]

is_connected()[source]

Return type:: bool

publish_once()[source]

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

Return type:: bool

save_snapshot(file_path)[source]

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

Parameters:: file_path (str)
Return type:: Dict[str, Any]

set_camera_config(width=None, height=None, fps=None)[source]

Parameters:

width (int | None)
height (int | None)
fps (float | None)

Return type:

Dict[str, Any]

set_camera_id(camera_id)[source]

Parameters:: camera_id (int)
Return type:: bool

set_stream_config(interval_sec=None, jpeg_quality=None, topic_name=None)[source]

gRPC配信設定を変更する。

Parameters:

interval_sec (float | None)
jpeg_quality (int | None)
topic_name (str | None)

Return type:

Dict[str, Any]

start_stream()[source]

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

Return type:: bool

stop_stream()[source]

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

Return type:: bool

Submodules

image_server.server_grpc.client module

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

class ImageStreamClient(*args, **kwargs)[source]

Bases: SyncGrpcClient

ImageStreamGrpcService用同期クライアント。

static decode_payload(payload)[source]

Parameters:: payload (Dict[str, Any])
Return type:: Any

iter_frames(topic='camera.frame')[source]

Parameters:: topic (str)
Return type:: Iterator[Tuple[Dict[str, Any], Any]]

show_camera_frame_stream(topic='camera.frame', window_name='camera.frame')[source]

Parameters:

topic (str)
window_name (str)

Return type:

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)[source]

Bases: object

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

_encode_frame(frame)[source]

Parameters:: frame (Any)
Return type:: Dict[str, Any]

_publish_loop()[source]

Return type:: None

_set_event_bus(event_bus)[source]

grpc_frameからEventBusを設定する。

Parameters:: event_bus (Any)
Return type:: None

static _time_ns()[source]

Return type:: int

static _validate_jpeg_quality(value)[source]

Parameters:: value (int)
Return type:: int

capture_once()[source]

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

Return type:: Dict[str, Any]

clear_error()[source]

Return type:: bool

connect()[source]

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

Return type:: bool

disconnect()[source]

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

Return type:: bool

get_config()[source]

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

Return type:: Dict[str, Any]

get_status()[source]

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

Return type:: Dict[str, Any]

is_connected()[source]

Return type:: bool

publish_once()[source]

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

Return type:: bool

save_snapshot(file_path)[source]

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

Parameters:: file_path (str)
Return type:: Dict[str, Any]

set_camera_config(width=None, height=None, fps=None)[source]

Parameters:

width (int | None)
height (int | None)
fps (float | None)

Return type:

Dict[str, Any]

set_camera_id(camera_id)[source]

Parameters:: camera_id (int)
Return type:: bool

set_stream_config(interval_sec=None, jpeg_quality=None, topic_name=None)[source]

gRPC配信設定を変更する。

Parameters:

interval_sec (float | None)
jpeg_quality (int | None)
topic_name (str | None)

Return type:

Dict[str, Any]

start_stream()[source]

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

Return type:: bool

stop_stream()[source]

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

Return type:: bool

image_server.server_grpc.make_pyi module

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

main()[source]: grpc_frameのgenerate_client_pyi()でclient.pyiを生成する。