Positioning: This guide introduces ese774 first because of SUSANOO’s BL774 origins, not because other Frames are technically subordinate. Begin with dynamic dispatch; use explicit API definitions only when needed. See the common development guide.
This is a development workflow for ese774 Frame, not a device-specific operating manual. Source excerpts illustrate each engineering step. Keep the DeviceClass independent of transport and separate server-side from client-side behavior. Installation is covered by the separate installation manual.
ese774_dummy/
├── dummy_device/
│ ├── dummy_ctrl.py
│ ├── dummy_router.py
│ ├── spec.py
│ ├── sync_dummy_client.py
│ └── async_dummy_client.py
├── dynamic/
│ ├── server.py
│ ├── client_sync.py
│ └── client_async.py
├── dynamic_router/
│ ├── server.py
│ ├── client_async.py
│ └── client_sync.py
├── spec/
│ ├── server.py
│ └── client_sync.py
└── local/
└── local_device_example.py
dummy_device/ holds the shared DeviceClass and the extensions at the transport boundary. dynamic/ demonstrates automatic dispatch, dynamic_router/ adds a custom Router, spec/ uses Pydantic and ApiSpec, and local/ illustrates direct use without networking. The DeviceClass is not reimplemented for each transport.
dummy_device/dummy_ctrl.pyfrom pathlib import Path
class DummyDeviceCtrl:
"""ese774_frame の手動統合試験用ダミーデバイス。"""
def __init__(self, initial_value: int = 10, logger=None, log_level=None, **kwargs):
self._value = int(initial_value)
self._scale = 1.0
self._logger = logger
self._files = {
"hello.txt": b"hello from dummy device\n",
"data.bin": bytes(range(16)),
}
def disconnect(self):
return None
def ping(self) -> str:
return "pong"
def add(self, a: int, b: int) -> int:
return a + b
def mixed(self, a: int, b: float, text: str = "") -> float:
return float(a) + float(b)
def get_complex_result(self):
return {"tuple": (1,2,3), "bytes": b"abc", "nested": {"value": self._value}}
def dangerous_reset(self):
self._value = 0
return True
def _private_api(self):
return "private"
@property
def value(self) -> int:
return self._value
@value.setter
def value(self, value: int):
self._value = int(value)
@property
def status(self) -> str:
return "connected"
def get_scale(self) -> float:
return self._scale
def set_scale(self, value: float):
self._scale = float(value)
scale = property(get_scale, set_scale)
def get_file_content(self, filename: str):
if filename not in self._files:
raise FileNotFoundError(filename)
return {"filename": filename, "content": self._files[filename]}
def download_file(self, filename: str, output_dir: str):
"""ローカル実行時はDeviceを実行しているPCへ保存する。"""
data = self.get_file_content(filename)
out = Path(output_dir)
out.mkdir(parents=True, exist_ok=True)
path = out / filename
path.write_bytes(data["content"])
return str(path)
ping(), add(), value, and status are ordinary Python interfaces. download_file() saves a file on the computer executing the DeviceClass. To save on the client instead, the Router and client override below adapt behavior across the network boundary. dangerous_reset() exists but need not be exposed.
dynamic/server.pyfrom pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
import os
from ese774_frame.api_server import FastApiServer
from dummy_device.dummy_ctrl import DummyDeviceCtrl
class Config:
HOST='127.0.0.1'
PORT=int(os.environ.get('ESE774_DUMMY_PORT','8000'))
server=FastApiServer(device_cls=DummyDeviceCtrl, router_cls=None, config=Config, api_spec=None, object_name='dummy', dispatch_exclude={'dangerous_reset'}, device_kwargs={'initial_value':10}, lifespan_msg_prefix='DUMMY')
if __name__=='__main__':
server.run(host=Config.HOST, port=Config.PORT)
api_spec=None and router_cls=None select automatic dispatch. dispatch_exclude={'dangerous_reset'} excludes an existing DeviceClass operation from remote exposure. device_kwargs supplies DeviceClass construction arguments without introducing transport concepts into the DeviceClass.
dynamic/client_sync.pyfrom pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
import os
PORT=int(os.environ.get('ESE774_DUMMY_PORT','8000'))
from ese774_frame.clients.sync_device_client import SyncDeviceClient
c=SyncDeviceClient(server_ip='127.0.0.1',server_port=PORT,object_name='dummy',api_spec=None)
try:
assert c.ping()=='pong'
assert c.add(2,5)==7
assert c.mixed(1,1.5,'x')==2.5
z=c.get_complex_result(); assert z['tuple']==(1,2,3) and z['bytes']==b'abc'
c.value=25; assert c.value==25
assert c.status=='connected'
c.scale=3.0; assert c.scale==3.0
print('DYNAMIC_SYNC_OK')
finally: c.close()
dynamic/client_async.pyfrom pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
import os
PORT=int(os.environ.get('ESE774_DUMMY_PORT','8000'))
import asyncio
from ese774_frame.clients.async_device_client import AsyncDeviceClient
async def main():
c=AsyncDeviceClient(server_ip='127.0.0.1',server_port=PORT,object_name='dummy',api_spec=None)
try:
assert await c.ping()=='pong'
assert await c.add(2,5)==7
c.value=40; assert c.value==40
assert c.status=='connected'
c.scale=5.0; assert c.scale==5.0
print('DYNAMIC_ASYNC_OK')
finally: await c.close()
if __name__=='__main__': asyncio.run(main())
Methods and readable/writable properties retain their DeviceClass-facing names. Synchronous and asynchronous clients differ in invocation, not in how the device functionality is defined.
dummy_device/dummy_router.pyfrom ese774_frame.routers.device_router import DeviceRouter
class DummyRouterCtrl(DeviceRouter):
"""サーバー/クライアントで意味が変わるAPIだけ override。"""
def download_file(self, filename: str, output_dir: str):
data = self._device.get_file_content(filename)
return {"filename": data["filename"], "content": data["content"], "output_dir": output_dir}
@property
def router_name(self) -> str:
return self.__class__.__name__
DummyRouterCtrl(DeviceRouter) returns file contents instead of interpreting a client-side output directory on the server. This does not make the DeviceClass an HTTP-specific subclass; it adapts an operation whose semantics change across the transport boundary.
dummy_device/sync_dummy_client.pyfrom pathlib import Path
from ese774_frame.clients.sync_device_client import SyncDeviceClient
class SyncDummyClient(SyncDeviceClient):
"""remote semantics が異なるAPIだけ client側処理を override。"""
def download_file(self, filename: str, output_dir: str):
if self.__dict__.get("_auto_dispatch", False):
result = self.dispatch("download_file", filename, output_dir)
else:
raw = getattr(self, "_download_file_raw")
result = raw(filename, output_dir)
path = Path(output_dir) / result["filename"]
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(result["content"])
return str(path)
dummy_device/async_dummy_client.pyfrom pathlib import Path
from ese774_frame.clients.async_device_client import AsyncDeviceClient
class AsyncDummyClient(AsyncDeviceClient):
"""remote semantics が異なるAPIだけ client側処理を override。"""
async def download_file(self, filename: str, output_dir: str):
if self.__dict__.get("_auto_dispatch", False):
result = await self.dispatch("download_file", filename, output_dir)
else:
raw = getattr(self, "_download_file_raw")
result = await raw(filename, output_dir)
path = Path(output_dir) / result["filename"]
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(result["content"])
return str(path)
The client-side download_file() override writes the received content to the client’s filesystem. The Router transformation and client override are a matched pair. _auto_dispatch and _download_file_raw appear in the supplied example as internal invocation details; they are not public DeviceClass interfaces or portable contracts across Frames.
dynamic_router/client_async.pyfrom pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
import os
PORT=int(os.environ.get('ESE774_DUMMY_PORT','8000'))
import asyncio
from dummy_device.async_dummy_client import AsyncDummyClient
async def main():
c=AsyncDummyClient(server_ip='127.0.0.1',server_port=PORT,object_name='dummy',api_spec=None)
try:
assert await c.ping()=='pong'; assert await c.add(2,5)==7
c.value=40; assert c.value==40
assert c.router_name=='DummyRouterCtrl'
p=await c.download_file('data.bin','client_download_async')
assert Path(p).read_bytes()==bytes(range(16))
print('DYNAMIC_ROUTER_ASYNC_OK')
finally: await c.close()
if __name__=='__main__': asyncio.run(main())
Here the public download_file() operation means “save on the client and return the resulting path.” The paired extensions preserve that behavior without making the DeviceClass depend on the network.
dummy_device/spec.pyfrom pydantic import BaseModel
from ese774_frame.models.api_spec import ApiSpec
class AddRequest(BaseModel):
a: int
b: int
class AddResponse(BaseModel):
result: int
class PingResponse(BaseModel):
result: str
class DownloadRequest(BaseModel):
filename: str
output_dir: str
class DownloadResponse(BaseModel):
filename: str
content: bytes
output_dir: str
DUMMY_API_SPEC = [
ApiSpec(request_model=None, response_model=PingResponse, name='ping', object_name='dummy', method='post'),
ApiSpec(request_model=AddRequest, response_model=AddResponse, name='add', object_name='dummy', method='post'),
ApiSpec(request_model=None, response_model=None, name='value', object_name='dummy', method='get', kind='property', writable=True),
ApiSpec(request_model=None, response_model=None, name='status', object_name='dummy', method='get', kind='property', writable=False),
ApiSpec(request_model=None, response_model=None, name='scale', object_name='dummy', method='get', kind='property', writable=True),
ApiSpec(request_model=DownloadRequest, response_model=DownloadResponse, name='download_file', object_name='dummy', method='post'),
]
DUMMY_API_SPEC enumerates exposed methods and properties and explicitly describes request and response models. This changes how the transport API is published, not the DeviceClass design. The example repository also contains spec/server.py and spec/client_sync.py, but their contents were not among the supplied files; this guide therefore does not invent their arguments or behavior.
The UR implementation demonstrates that the same development pattern applies beyond COBOTTA. The tree below represents the source package under src/ur/. build/lib/ur/ is a build artifact, not the source tree; docs/, examples/, and tests/ are repository-level siblings of src/.
ur/
├── pyproject.toml
├── src/
│ └── ur/
│ ├── __init__.py
│ ├── ur_ctrl.py # DeviceClass implementation
│ ├── config.py
│ └── server_fastapi/
│ ├── __init__.py # Exports and DeviceProxy registration
│ ├── spec_ctrl.py # ApiSpec declarations
│ ├── make.py # .pyi generation
│ ├── models/
│ │ ├── common.py
│ │ ├── motion.py
│ │ └── ...
│ ├── routers/
│ │ └── ur_router_ctrl.py
│ └── clients/
│ ├── sync_ur_client.py
│ └── async_ur_client.py
├── examples/
│ ├── run_server.py
│ ├── client_sample.py
│ └── ur_server.yaml
└── docs/
├── ja/
└── en/
server_fastapi/ belongs inside the installable ur package, not under examples/. The DeviceClass and its server/client adapters can therefore be imported from the same installed package. Pydantic request/response models live under models/; spec_ctrl.py declares the exposed API; routers/ and clients/ are separate extension points.
The supplied examples/run_server.py uses this construction (configuration loading and logging setup omitted):
from ur import Config, UrCtrl
from ur.server_fastapi import UrRouterCtrl, ur_ctrl_api_spec
from ese774_frame import FastApiServer
server = FastApiServer(
device_cls=UrCtrl,
router_cls=UrRouterCtrl,
config=Config,
api_spec=ur_ctrl_api_spec,
logger=logger,
logger_name=Config.SERVER_LOGGER_NAME,
lifespan_msg_prefix="UR",
)
server.run(host=Config.SERVER_IP, port=Config.SERVER_PORT)
device_cls=UrCtrl selects the existing DeviceClass, without creating a server-specific subclass. UrRouterCtrl(DeviceRouter) is a thin wrapper forwarding device_instance, api_spec, logger, and log_level to its base class; it does not implement robot control. This keeps the device implementation separate from the HTTP boundary.
spec_ctrl.py imports the models from server_fastapi/models/ and declares ur_ctrl_api_spec. Two entries from the actual file are shown here:
from ese774_frame.models.api_spec import ApiSpec
from ur.server_fastapi.models.common import ResultResponse, VersionRequest
ur_ctrl_api_spec = [
ApiSpec(
name="connect", object_name="ur",
request_model=None, response_model=ResultResponse,
method="post", description="connect()",
summary="UR コントローラへ接続する",
),
ApiSpec(
name="get_version", object_name="ur",
request_model=VersionRequest, response_model=ResultResponse,
method="post", description="get_version(rel_toml_path=None)",
summary="UR 制御環境のバージョン情報を取得する",
),
]
This is an excerpt, not the complete specification. Explicit APIs define the published operations and request/response types at the server boundary; they do not make the DeviceClass transport-dependent. Automatic dispatch is covered earlier in this article.
SyncURClient(SyncDeviceClient) and AsyncURClient(AsyncDeviceClient) pass the endpoint and ur_ctrl_api_spec to the framework client. Normal device operations are not reimplemented on the client. The synchronous client’s setup follows this structure:
from ese774_frame.clients.sync_device_client import SyncDeviceClient
from ur.server_fastapi.spec_ctrl import ur_ctrl_api_spec
class SyncURClient(SyncDeviceClient):
def __init__(self, server_ip=None, server_port=None, logger=None, **kwargs):
# Illustrative excerpt; the actual constructor also accepts config/log_level.
server_ip = server_ip or "127.0.0.1"
server_port = int(server_port or 8000)
super().__init__(
server_ip=server_ip, server_port=server_port,
api_spec=ur_ctrl_api_spec,
base_url=f"http://{server_ip}:{server_port}",
logger=logger,
)
This is an illustrative excerpt, not a verbatim reproduction of the constructor. The actual asynchronous client also implements __getattr__ and __setattr__ to map P1/J1/F1/I1/IO1-style access to get_var/set_var. It also handles disconnect/delete on the client so that closing a client does not inadvertently disconnect the server-side device. These are client-facing adaptations, not a relocation of device-control responsibilities.
.pyi separately from implementationsserver_fastapi/make.py calls make_pyi_device_client, make_pyi_device_router, and make_pyi_device_proxy to generate clients/*.pyi, routers/*.pyi, and server_fastapi/__init__.pyi. It does not overwrite the implementation .py files. The proxy stub generation uses device_class="UrRtde", aliases=["ur"], and the sync/async client class names. Type stubs support IDE completion and static checking; they are not communication implementations.
The reusable layout illustrated by UR is: place the DeviceClass under src/<package>/, keep the framework integration under src/<package>/server_fastapi/, put runnable examples at repository-level examples/, and keep Routers, clients, API models/specifications, and type stubs distinct. This is an observed, reusable organization, not a mandatory framework directory naming rule.
In aandd_reader, src/aandd_reader/aandd_reader.py contains the DeviceClass, server_fastapi/ the HTTP adapter, server_grpc/ the gRPC adapter, and examples/ the launch and usage scripts. Supporting both transports does not require duplicating the DeviceClass.
Core excerpt from examples/fastapi_server.py (Config and logger are initialized elsewhere in that script):
server = FastApiServer(
device_cls=AanddReader,
router_cls=AanddReaderRouter,
config=Config,
api_spec=aandd_reader_api_spec,
device_kwargs={"port": Config.DEVICE_PORT},
logger=logger,
logger_name=Config.SERVER_LOGGER_NAME,
lifespan_msg_prefix="AANDD",
)
server.run(host=Config.SERVER_IP, port=Config.FASTAPI_PORT)
device_cls supplies the class, device_kwargs its constructor arguments, router_cls the server-side extension, and api_spec the explicit published interface. None belongs inside the DeviceClass.
AanddReaderRouter(DeviceRouter) forwards device_instance and api_spec to super().__init__; overriding device behavior in every router is not necessary. The supplied model file is literally named server_fastapi/models/comman.py:
from typing import Any, Optional
from pydantic import BaseModel
class ResultResponse(BaseModel):
result: Optional[Any]
class SetMeasModeRequest(BaseModel):
mode: str
spec_ctrl.py maps set_meas_mode to SetMeasModeRequest and ResultResponse using ApiSpec(..., object_name="aandd", method="post", ...). This is explicit API mode; automatic dispatch does not require these models.
server_fastapi/__init__.py registers the name AanddReader with register_device_proxy, specifying both custom clients, api_spec, default_async_mode=True, and aliases. examples/fastapi_device_proxy.py imports DeviceProxy from that package, constructs DeviceProxy("AanddReader", config=Config, logger=logger, log_level=log_level, async_mode=True), then calls await client.meas().
make.py generates client/router .pyi files and post-processes them to preserve the actual class inheritance and helper methods; make_device_proxy.py generates __init__.pyi. Typing declarations must match the real implementation. The matching gRPC adapter is covered in the gRPC Frame guide.
COBOTTA uses the same separation through server_fastapi/models/, routers/, clients/, and spec_ctrl.py; see the development guide for its package layout and actual launch and control scripts. gRPC also separates DeviceClass from transport, but the Pydantic / ApiSpec approach shown here is specific to ese774 Frame; gRPC uses automatic dispatch only. A concrete gRPC example must be grounded in its own source code.