本指南包含一个教程，介绍了如何编写脚本来远程与 Fuchsia 设备交互，以及各种其他脚本化交互示例，还包含一个关于在主机上使用 Python 实现 FIDL 服务器的入门介绍。
本页介绍的方法最初是为端到端测试而开发的，但您可能还会发现，它们在编写简单的设备互动脚本时也非常有用，而无需在 Rust 中编写 ffx 插件。
背景
以下技术支持通过脚本与 Fuchsia 设备进行远程互动：
Fuchsia Controller 和 Overnet 可在运行 Python 脚本的主机和目标 Fuchsia 设备之间提供传输。Python FIDL 绑定（与其他语言中的 FIDL 绑定类似）有助于向目标设备上的组件发送和接收消息。
通常，您只需在连接到目标设备后与 Python FIDL 绑定进行交互。如您所见，连接到目标设备是使用 fuchsia_controller_py 模块中的 Context 类完成的。
除了 Overnet 之外，libfuchsia_controller_internal.so 共享库（包含 ABI 头文件）是驱动脚本化远程交互的核心库。
教程
本教程将逐步介绍如何编写一个 Python 脚本，用于读取地址列表并输出每个地址对应的 Fuchsia 设备的相关信息。
如果您遇到 bug 或有疑问或建议，请提交问题。
前提条件
本教程需要满足以下条件：
- Fuchsia 源代码检出和关联的开发环境。
可通过公开远程控制服务 (RCS) 的
ffx访问的 Fuchsia 设备（实体或模拟设备）。运行
ffx target list时，RCS下的字段必须显示Y：NAME SERIAL TYPE STATE ADDRS/IP RCS fuchsia-emulator <unknown> Unknown Product [fe80::5054:ff:fe63:5e7a%4] Y（如需了解详情，请参阅与目标设备互动。）
创建 build 目标
更新 BUILD.gn 文件，使其包含如下所示的 build 目标：
import("//build/python/python_binary.gni")
assert(is_host)
python_binary("describe_host_example") {
testonly = true
main_source = "examples/describe_host.py"
deps = [
"//sdk/fidl/fuchsia.buildinfo:fuchsia.buildinfo_python",
"//sdk/fidl/fuchsia.developer.remotecontrol:fuchsia.developer.remotecontrol_python",
"//src/developer/ffx:host",
]
}
ffx 工具可让 ffx 守护程序连接到我们的 Fuchsia 设备。FIDL 依赖项可将必要的 Python FIDL 绑定提供给脚本。
编写脚本
首先，我们的脚本需要导入必要的模块：
import fidl_fuchsia_buildinfo as f_buildinfo
import fidl_fuchsia_developer_remotecontrol as f_remote_control
from fuchsia_controller_py import Context
fidl_fuchsia_developer_remotecontrol 和 fidl_fuchsia_buildinfo Python 模块包含 fuchsia.developer.ffx 和 fuchsia.buildinfo FIDL 库的 Python FIDL 绑定。
Context 对象可提供 ffx 守护程序和 Fuchsia 目标之间的连接。
接下来，我们的脚本需要定义一个函数（在此示例中称为 describe_host），以从目标设备检索信息：
# Return information about a Fuchsia target.
async def describe_host(target_ip: str) -> str:
ctx = Context(
target=target_ip,
)
remote_control_proxy = f_remote_control.RemoteControlClient(
ctx.connect_remote_control_proxy()
)
identify_host_response = (
# The `.response` in the next line is a bit confusing. The .unwrap() extracts the
# RemoteControlIdentifyHostResponse from the response field of the returned
# RemoteControlIdentifyHostResult, but RemoteControlIdentifyHostResponse also contains a
# response field. There are two nested response fields.
(await remote_control_proxy.identify_host())
.unwrap()
.response
)
build_info_proxy = f_buildinfo.ProviderClient(
ctx.connect_device_proxy(
"/core/build-info", f_buildinfo.ProviderMarker
)
)
buildinfo_response = await build_info_proxy.get_build_info()
return f"""
--- {target_ip} ---
nodename: {identify_host_response.nodename}
product_config: {buildinfo_response.build_info.product_config}
board_config: {buildinfo_response.build_info.board_config}
version: {buildinfo_response.build_info.version}
"""
此函数会实例化 Context 以连接到特定 IP 地址的 Fuchsia 设备，然后调用 fuchsia.developer.remotecontrol/RemoteControl.IdentifyHost 和 fuchsia.buildinfo/Provider.GetBuildInfo 方法来获取目标的相关信息。
最后，您需要使用一些 Python 样板代码封装此代码，以读取地址并输出每个目标设备的信息。因此，您会得到以下脚本：
import argparse
import asyncio
import fidl_fuchsia_buildinfo as f_buildinfo
import fidl_fuchsia_developer_remotecontrol as f_remote_control
from fuchsia_controller_py import Context
USAGE = "This script prints the result of RemoteControl.IdentifyHost on Fuchsia targets."
async def main() -> None:
params = argparse.ArgumentParser(usage=USAGE)
params.add_argument(
"target_ips",
help="Fuchsia target IP addresses",
nargs="+",
)
args = params.parse_args()
# Return information about a Fuchsia target.
async def describe_host(target_ip: str) -> str:
ctx = Context(
target=target_ip,
)
remote_control_proxy = f_remote_control.RemoteControlClient(
ctx.connect_remote_control_proxy()
)
identify_host_response = (
# The `.response` in the next line is a bit confusing. The .unwrap() extracts the
# RemoteControlIdentifyHostResponse from the response field of the returned
# RemoteControlIdentifyHostResult, but RemoteControlIdentifyHostResponse also contains a
# response field. There are two nested response fields.
(await remote_control_proxy.identify_host())
.unwrap()
.response
)
build_info_proxy = f_buildinfo.ProviderClient(
ctx.connect_device_proxy(
"/core/build-info", f_buildinfo.ProviderMarker
)
)
buildinfo_response = await build_info_proxy.get_build_info()
return f"""
--- {target_ip} ---
nodename: {identify_host_response.nodename}
product_config: {buildinfo_response.build_info.product_config}
board_config: {buildinfo_response.build_info.board_config}
version: {buildinfo_response.build_info.version}
"""
# Asynchronously await information from each Fuchsia target.
results = await asyncio.gather(*map(describe_host, args.target_ips))
print("Target Info Received:")
print("\n".join(results))
构建并运行脚本
此示例代码位于 //tools/fidl/fidlgen_python 中，因此您可以在 x64 主机上使用以下命令构建该代码（在使用 fx args 将 //tools/fidl/fidlgen_python:examples 添加到 host_labels 后）：
fx build --host //tools/fidl/fidlgen_python:describe_host_example接下来，您将使用 ffx target list 来识别目标设备的地址。
$ ffx target list
NAME SERIAL TYPE STATE ADDRS/IP RCS
fuchsia-emulator <unknown> core.x64 Product [127.0.0.1:34953] Y
然后，运行脚本！
$ fx run-in-build-dir host_x64/obj/tools/fidl/fidlgen_python/describe_host_example.pyz '127.0.0.1:34953'
Target Info Received:
--- 127.0.0.1:34953 ---
nodename: fuchsia-emulator
product_config: core
board_config: x64
version: 2025-04-08T02:04:13+00:00
查找组件标识名
若要与 Fuchsia 组件通信，脚本必须事先知道组件的别名。您可以使用 ffx 检索组件标识名。例如，以下 ffx 命令将输出 core/build-info 公开了 fuchsia.buildinfo/Provider capability：
ffx component capability fuchsia.buildinfo.Provider此命令将输出类似于以下内容的输出：
Declarations:
`core/build-info` declared capability `fuchsia.buildinfo.Provider`
Exposes:
`core/build-info` exposed `fuchsia.buildinfo.Provider` from self to parent
Offers:
`core` offered `fuchsia.buildinfo.Provider` from child `#build-info` to child `#cobalt`
`core` offered `fuchsia.buildinfo.Provider` from child `#build-info` to child `#remote-control`
`core` offered `fuchsia.buildinfo.Provider` from child `#build-info` to child `#sshd-host`
`core` offered `fuchsia.buildinfo.Provider` from child `#build-info` to child `#test_manager`
`core` offered `fuchsia.buildinfo.Provider` from child `#build-info` to child `#testing`
`core` offered `fuchsia.buildinfo.Provider` from child `#build-info` to child `#toolbox`
`core/sshd-host` offered `fuchsia.buildinfo.Provider` from parent to collection `#shell`
Uses:
`core/remote-control` used `fuchsia.buildinfo.Provider` from parent
`core/sshd-host/shell:sshd-0` used `fuchsia.buildinfo.Provider` from parent
`core/cobalt` used `fuchsia.buildinfo.Provider` from parent
其他示例
本部分演示了各种其他脚本化互动。
重启设备
您可以通过多种方式重新启动设备。重新启动设备的方法之一是连接到运行 fuchsia.hardware.power.statecontrol/Admin 协议的组件，该协议可在 /bootstrap/shutdown_shim 下找到。
采用这种方法时，协议预计会在方法执行期间退出并返回 PEER_CLOSED 错误：
ch = self.device.ctx.connect_device_proxy(
"bootstrap/shutdown_shim", power_statecontrol.AdminMarker
)
admin = power_statecontrol.AdminClient(ch)
# Makes a coroutine to ensure that a PEER_CLOSED isn't received from attempting
# to write to the channel.
coro = admin.reboot(reason=power_statecontrol.RebootReason.USER_REQUEST)
try:
_LOGGER.info("Issuing reboot command")
await coro
except ZxStatus as status:
if status.raw() != ZxStatus.ZX_ERR_PEER_CLOSED:
raise status
_LOGGER.info("Device reboot command sent")
不过，接下来需要确定设备是否已重新上线，这部分比较棘手。通常，这通过尝试连接到某个协议（通常是 RemoteControl 协议）直到超时来实现。
另一种方法（可减少代码量）是连接到 ffx 守护程序的 Target 协议：
import fidl_fuchsia_developer_ffx as f_ffx
ch = ctx.connect_target_proxy()
target_proxy = f_ffx.TargetClient(ch)
await target_proxy.reboot(state=f_ffx.TargetRebootState.PRODUCT)
运行组件
您可以使用 RemoteControl 协议启动组件，具体步骤如下：
连接到生命周期控制器：
import fidl_fuchsia_developer_remotecontrol as f_remotecontrol import fidl_fuchsia_sys2 as f_sys2 ch = ctx.connect_to_remote_control_proxy() remote_control_proxy = f_remotecontrol.RemoteControlClient(ch) client, server = fuchsia_controller_py.Channel.create() await remote_control_proxy.root_lifecycle_controller(server=server.take()) lifecycle_ctrl = f_sys2.LifecycleControllerClient(client)尝试启动组件实例：
import fidl_fuchsia_component as f_component client, server = fuchsia_controller_py.Channel.create() await lifecycle_ctrl.start_instance("some_moniker", server=server.take()) binder = f_component.BinderClient(client)binder对象可让用户知道组件是否保持连接。不过，它没有方法。尚未实现用于确定组件是否已解除绑定（使用 binder 协议）的支持。
获取快照
从 Fuchsia 设备获取快照需要运行快照并绑定 File 协议以进行读取：
client, server = Channel.create()
file = io.FileClient(client)
params = feedback.GetSnapshotParameters(
# Two minutes of timeout time.
collection_timeout_per_data=(2 * 60 * 10**9),
response_channel=server.take(),
)
assert self.device.ctx is not None
ch = self.device.ctx.connect_device_proxy(
"/core/feedback", "fuchsia.feedback.DataProvider"
)
provider = feedback.DataProviderClient(ch)
await provider.get_snapshot(params=params)
attr_res = await file.get_attr()
asserts.assert_equal(attr_res.s, ZxStatus.ZX_OK)
data = bytearray()
while True:
response = (await file.read(count=io.MAX_BUF)).unwrap()
if not response.data:
break
data.extend(response.data)
在主机上实现 FIDL 服务器
Fuchsia Controller 的一项重要任务（用于处理传递的绑定或测试复杂的客户端代码）是运行 FIDL 服务器。在本部分中，您将返回 echo 示例并实现 echo 服务器。您需要替换的函数派生自 FIDL 文件定义。因此，echo 服务器（使用 ffx 协议）将如下所示：
class TestEchoer(ffx.EchoServer):
def echo_string(
self, request: ffx.EchoEchoStringRequest
) -> ffx.EchoEchoStringResponse:
return ffx.EchoEchoStringResponse(response=request.value)
如需进行适当的实现，您需要导入适当的库。与之前一样，您需要导入 fidl_fuchsia_developer_ffx。不过，由于您要运行 echo 服务器，因此测试此服务器的最快方法是使用 fuchsia_controller_py 库中的 Channel 对象：
import fidl_fuchsia_developer_ffx as ffx
from fuchsia_controller_py import Channel
此 Channel 对象的行为与其他语言中的对象类似。以下代码是一个利用 echo 服务器的简单程序：
import asyncio
import unittest
import fidl_fuchsia_developer_ffx as ffx
from fuchsia_controller_py import Channel
class TestEchoer(ffx.EchoServer):
def echo_string(
self, request: ffx.EchoEchoStringRequest
) -> ffx.EchoEchoStringResponse:
return ffx.EchoEchoStringResponse(response=request.value)
class TestCases(unittest.IsolatedAsyncioTestCase):
async def test_echoer_example(self):
(tx, rx) = Channel.create()
server = TestEchoer(rx)
client = ffx.EchoClient(tx)
server_task = asyncio.get_running_loop().create_task(server.serve())
res = await client.echo_string(value="foobar")
self.assertEqual(res.response, "foobar")
server_task.cancel()
实现服务器时，需要注意以下几点：
- 方法定义可以是
sync或async。 serve()任务将处理请求并调用服务器实现中的必要方法，直到任务完成或底层通道对象关闭为止。- 如果在投放任务运行时发生异常，客户端渠道会收到
PEER_CLOSED错误。然后，您必须检查投放任务的结果。 - 与 Rust 的异步代码不同，在创建异步任务时，您必须保留返回的对象，直到完成使用为止。否则，系统可能会对任务进行垃圾回收并将其取消。
常见的 FIDL 服务器代码模式
与上面的简单 echo 服务器示例相比，本部分介绍了不同类型的服务器互动。
创建 FIDL 服务器类
我们将使用以下 FIDL 协议来创建服务器：
library fuchsia.exampleserver;
type SomeGenericError = flexible enum {
THIS = 1;
THAT = 2;
THOSE = 3;
};
closed protocol Example {
strict CheckFileExists(struct {
path string:255;
follow_symlinks bool;
}) -> (struct {
exists bool;
}) error SomeGenericError;
};
FIDL 方法名称的派生方式是将方法名称从驼峰式大小写法更改为小写蛇形命名法。因此，Python 中的 CheckFileExists 方法会更改为 check_file_exists。
匿名结构体类型派生自完整的协议名称和方法。因此，它们可能会非常详尽。输入法的输入参数定义为名为 ExampleCheckFileExistsRequest 的类型。响应称为 ExampleCheckFileExistsResponse。
将这些内容整合起来，Python 中的 FIDL 服务器实现如下所示：
import fidl_fuchsia_exampleserver as fe
class ExampleServerImpl(fe.ExampleServer):
def some_file_check_function(path: str) -> bool:
# Just pretend a real check happens here.
return True
def check_file_exists(self, req: fe.ExampleCheckFileExistsRequest) -> fe.ExampleCheckFileExistsResponse:
return fe.ExampleCheckFileExistsResponse(
exists=ExampleServerImpl.some_file_check_function()
)
您也可以将这些方法作为 async 实现，而不会出现任何问题。
此外，如需返回错误，需要将错误封装在 FIDL DomainError 对象中，例如：
import fidl_fuchsia_exampleserver as fe
from fidl import DomainError
class ExampleServerImpl(fe.ExampleServer):
def check_file_exists(self, req: fe.ExampleCheckFileExistsRequests) -> fe.ExampleCheckFileExistsResponse | DomainError:
return DomainError(error=fe.SomeGenericError.THIS)
处理事件
事件处理程序的编写方式与服务器类似。事件在渠道的客户端处理，因此必须传递客户端才能构建事件处理脚本。
我们先从以下 FIDL 代码开始构建一个示例：
library fuchsia.exampleserver;
closed protocol Example {
strict -> OnFirst(struct {
message string:128;
});
strict -> OnSecond();
};
此 FIDL 示例包含事件处理脚本需要处理的两个不同的事件。编写仅执行输出的最简单类如下所示：
import fidl_fuchsia_exampleserver as fe
class ExampleEventHandler(fe.ExampleEventHandler):
def on_first(self, req: fe.ExampleOnFirstRequest):
print(f"Got a message on first: {req.message}")
def on_second(self):
print(f"Got an 'on second' event")
如果您想在不出错的情况下停止处理事件，可以引发 fidl.StopEventHandler。
您可以使用一些现有的 fidlgen_python 测试代码来测试此事件示例。但首先，请确保已将 Fuchsia 控制器测试添加到 Fuchsia build 设置，例如：
fx set ... --with-host //tools/fidl/fidlgen_python:tests
使用 fuchsia.controller.test 中的协议（在 fuchsia_controller.test.fidl 中定义），您可以编写使用 ExampleEvents 协议的代码，例如：
import asyncio
import fidl_fuchsia_controller_test as fct
from fidl import StopEventHandler
from fuchsia_controller_py import Channel
class ExampleEventHandler(fct.ExampleEventsEventHandler):
def on_first(self, req: fct.ExampleEventsOnFirstRequest):
print(f"Got on-first event message: {req.message}")
def on_second(self):
print(f"Got on-second event")
raise StopEventHandler
async def main():
client_chan, server_chan = Channel.create()
client = fct.ExampleEventsClient(client_chan)
server = fct.ExampleEventsServer(server_chan)
event_handler = ExampleEventHandler(client)
event_handler_task = asyncio.get_running_loop().create_task(
event_handler.serve()
)
server.on_first(message="first message")
server.on_second()
server.on_complete()
await event_handler_task
if __name__ == "__main__":
asyncio.run(main())
然后，通过完成下一部分中的 Python 环境设置步骤，即可运行该脚本。运行时，该脚本会输出以下内容并退出：
Got on-first event message: first message
Got on-second event
如需查看有关服务器测试的更多示例，请参阅此 test_server_and_event_handler.py 文件。