ese774_frame.clients package

Submodules

ese774_frame.clients.async_device_client module

FastAPI 用の汎用 async クライアント。

• api_spec からメソッドを動的生成

• HTTP で device API を呼び出し

• adapter で args/kwargs/result を pack/unpack

Kengo NAKADA: https://github.com/shimane-dev, https://github.com/kengo-nakada kengo.nakada@mat.shimane-u.ac.jp, kengo.nakada@gmail.com

class ese774_frame.clients.async_device_client.AsyncDeviceClient(server_ip: str = '127.0.0.1', server_port: int = 8000, base_url: str | None = None, api_spec: list | None = None, logger: Any | None = None, log_level: str | None = None, object_name: str = 'device', timeout_sec: float = 60.0)

ベースクラス: object

api_spec 駆動の非同期クライアント。

役割: - 呼び出し引数を request_model に合わせて正規化・検証 - HTTP 呼び出し実行 - レスポンスを ApiSpec.decode_response（または互換復元）で

プレーン Python に戻して返す

• __dispatch__ 経路では adapter で args/kwargs/result を透過運搬

static auto_extract_result(obj: Any) → Any

BaseModel または dict を自動アンラップする。 0.3.8 互換: BaseModel/辞書の1フィールドを自動アンラップし、複数フィールドはdictで返す。

async close() → None

HTTP client を閉じる

async dispatch(method: str, *args, **kwargs) → Any

一般形ディスパッチ (*args, **kwargs) を実行する。

payload 形式:

{

"method": str, "args": packed args, "kwargs": packed kwargs,

}

local_echo(message: str = 'local method ok') → str

API spec や Pydantic を使わない、クライアントローカル確認用メソッド。

set_timeout(timeout_sec: float) → None

HTTP 通信の timeout 秒数を変更する。 次回以降の request から新しい timeout を使う。

ese774_frame.clients.device_proxy module

ese774_frame.clients.device_proxy.DeviceProxy(device_class: str, *args, async_mode: bool | None = None, **kwargs) → Any

device_class から client 実体を生成する。

async_mode:

True -> async client False -> sync client None -> register_device_proxy(..., default_async_mode=...) に従う

async_mode 以外の *args/**kwargs は client class へ完全転送する。

class ese774_frame.clients.device_proxy.DeviceProxyEntry(device_class: 'str', async_client_cls: 'Optional[Type[Any]]' = None, sync_client_cls: 'Optional[Type[Any]]' = None, api_spec: 'Optional[list]' = None, object_name: 'Optional[str]' = None, default_async_mode: 'bool' = True)

ベースクラス: object

api_spec: list | None = None

async_client_cls: Type[Any] | None = None

default_async_mode: bool = True

device_class: str

object_name: str | None = None

sync_client_cls: Type[Any] | None = None

ese774_frame.clients.device_proxy.create_device_proxy(device_class: str)

device_class を固定した DeviceProxy を作成する。

instance = DiceProxy("device_class") でinstance を作ると instance.pyi での補完が機能しない問題に対処しやすいように フレーム側でのサポート関数

パラメータ:

device_class -- 登録済み device class 名。

戻り値:

device_class を固定した proxy 関数。

ese774_frame.clients.device_proxy.get_device_proxy_entry(device_class: str) → DeviceProxyEntry

ese774_frame.clients.device_proxy.list_device_proxies() → List[str]

ese774_frame.clients.device_proxy.register_device_proxy(device_class: str, *, async_client_cls: Type[Any] | None = None, sync_client_cls: Type[Any] | None = None, api_spec: list | None = None, object_name: str | None = None, default_async_mode: bool = True, aliases: List[str] | None = None) → None

ese774_frame.clients.device_proxy.unregister_device_proxy(device_class: str) → None

ese774_frame.clients.make_pyi_device_client module

PYIスタブファイル自動生成 (device client: async / sync 共通)

Kengo NAKADA: https://github.com/shimane-dev, https://github.com/kengo-nakada kengo.nakada@mat.shimane-u.ac.jp, kengo.nakada@gmail.com

ese774_frame.clients.make_pyi_device_client.collect_type_hints_from_model(model)

ese774_frame.clients.make_pyi_device_client.extract_all_types(tp)

ese774_frame.clients.make_pyi_device_client.gen_api_method_signatures(api, *, async_mode: bool)

1 API 分の overload シグネチャを生成する。

生成順: - req: Model - 展開キーワード引数 - params: dict - no-arg - **kwargs - *args, **kwargs

ese774_frame.clients.make_pyi_device_client.gen_api_property_signatures(api)

ApiSpec(kind="property") 1件分の property stub を生成する。

ese774_frame.clients.make_pyi_device_client.gen_auto_method_signature(name: str, sig: Signature, *, async_mode: bool, drop_first: bool)

完全自動 dispatch の public method 1件分の stub を生成する。

ese774_frame.clients.make_pyi_device_client.gen_auto_property_signature(name: str, prop: property)

完全自動 dispatch の public property 1件分の stub を生成する。

ese774_frame.clients.make_pyi_device_client.make_pyi_device_client(*, filename: str, class_name: str, async_mode: bool, api_spec=None, device_class=None)

device client 用 pyi を生成する。

公開対象の定義方法は Framework の2方式に対応する。

• api_spec を指定:

  spec に定義された API から生成する。

• device_class を指定:

  完全自動 dispatch と同じく public method/property を device class から収集して生成する。

api_spec と device_class は同時には指定しない。

ese774_frame.clients.make_pyi_device_proxy module

DeviceProxy 用 pyi 生成器。

spec / 完全自動のどちらでも公開入口は make_pyi_device_proxy の1つ。 DeviceProxy の型付けに必要なのは最終的な sync/async client 型なので、 API の定義方法そのものはここでは分岐させない。

ese774_frame.clients.make_pyi_device_proxy.make_pyi_device_proxy(*, filename: str, device_names: Iterable[str], async_client_module: str, async_client_class: str, sync_client_module: str, sync_client_class: str, default_async_mode: bool = True, proxy_name: str = 'DeviceProxy', create_proxy_name: str | None = 'create_device_proxy') → None

DeviceProxy の戻り型を登録済み client 型へ結び付ける pyi を生成する。

api_spec で client pyi を生成した場合も、device_class から完全自動で client pyi を生成した場合も、この関数の呼び方は同じ。

ese774_frame.clients.sync_device_client module

FastAPI 用の汎用 sync クライアント。

• api_spec からメソッドを動的生成

• HTTP で device API を呼び出し

• adapter で args/kwargs/result を pack/unpack

Kengo NAKADA: https://github.com/shimane-dev, https://github.com/kengo-nakada kengo.nakada@mat.shimane-u.ac.jp, kengo.nakada@gmail.com

class ese774_frame.clients.sync_device_client.SyncDeviceClient(server_ip: str = '127.0.0.1', server_port: int = 8000, base_url: str | None = None, api_spec: list | None = None, logger: Any | None = None, log_level: str | None = None, object_name: str = 'device', timeout_sec: float = 60.0)

ベースクラス: object

api_spec 駆動の同期クライアント。

役割: - 呼び出し引数を request_model に合わせて正規化・検証 - 同期 HTTP 呼び出し実行 - レスポンスを ApiSpec.decode_response（または互換復元）で

プレーン Python に戻して返す

• __dispatch__ 経路では adapter で args/kwargs/result を透過運搬

static auto_extract_result(obj: Any) → Any

BaseModel または dict を自動アンラップする。

BaseModel/辞書の1フィールドを自動アンラップし、 複数フィールドは dict で返す。

close() → None

HTTP client を同期的に閉じる。

close_client() → None

既存の同期クライアント互換名で HTTP client を閉じる。

dispatch(method: str, *args, **kwargs) → Any

一般形ディスパッチ (*args, **kwargs) を同期実行する。

local_echo(message: str = 'local method ok') → str

API spec や Pydantic を使わない、クライアントローカル確認用メソッド。

set_timeout(timeout_sec: float) → None

HTTP 通信の timeout 秒数を変更する。

既存の同期 HTTP client を閉じ、 次回以降の request から新しい timeout を使う。