本教程详细介绍了如何在 Fuchsia 源检出 (fuchsia.git
) 设置中编写使用 Fuchsia Controller (fuchsia-controller
) 的简单脚本。
Fuchsia 控制器由一组库组成,可让用户连接到 Fuchsia 设备并使用 FIDL 与设备进行交互。Fuchsia 控制器最初是为测试而创建的。不过,它对于创建与 Fuchsia 设备上的 FIDL 接口进行交互的可脚本代码也很有用。例如,用户可以使用 Fuchsia Controller 编写脚本来执行简单的设备交互,而无需在 Rust 中编写 ffx
插件。
Fuchsia Controller 的主要两部分包括:
fuchsia-controller.so
文件(包含 ABI 的头文件)更高级别的语言绑定(基于使用 ABI 的
.so
文件构建)目前,Fuchsia Controller 的更高级别语言绑定仅使用 Python 编写。
使用 Fuchsia Controller 的最快方法是编写使用 fuchsia-controller
代码的 Python 脚本。在 Fuchsia 源检出设置中,您可以将 Python 二进制文件构建到 .pyz
文件中,然后可以从 out
目录(例如 $FUCHSIA_DIR/out/default
)执行该文件。
如需编写您的第一个 Fuchsia Controller 脚本,请按以下步骤操作:
如果您遇到 bug 或者有疑问或建议,请提交 bug。
前提条件
本教程要求您具备以下前提条件:
您需要使用 Fuchsia 源检出 (
fuchsia.git
) 开发环境。您需要运行一台 Fuchsia 设备。这可以是实体设备或模拟器。
此设备必须连接到
ffx
,并且已正确连接遥控器服务 (RCS)。如果运行
ffx target list
,则RCS
下的字段必须显示为Y
,例如:NAME SERIAL TYPE STATE ADDRS/IP RCS fuchsia-emulator <unknown> Unknown Product [fe80::5054:ff:fe63:5e7a%4] Y
(如需了解详情,请参阅与目标设备交互)。
如需在启用网络但不支持图形界面的情况下启动 Fuchsia 模拟器,请运行
ffx emu start --headless
。(如需了解详情,请参阅启动 Fuchsia 模拟器。)您的设备必须至少正在运行
core
产品。
更新 BUILD.gn 中的依赖项
更新 BUILD.gn
文件以包含以下依赖项:
import("//build/python/python_binary.gni")
assert(is_host)
python_binary("your_binary") {
main_source = "path/to/your/main.py"
deps = [
"//src/developer/ffx:host",
"//src/developer/ffx/lib/fuchsia-controller:fidl_bindings",
"//src/developer/ffx/lib/fuchsia-controller:fuchsia_controller_py",
]
}
fidl_bindings
规则包含必要的 Python 和 .so
绑定代码。还必须包含 ffx
工具,以便 ffx
守护程序能够连接到您的 Fuchsia 设备。
编写您的第一个程序
在本部分中,我们将创建一个简单的程序,该程序尚未连接到 Fuchsia 设备,但会连接到 ffx
守护程序以验证设备是否已启动并运行。为此,我们利用现有的 ffx
FIDL 库与在 //src/developer/ffx/fidl
中定义的守护程序进行交互。
添加 FIDL 依赖项
Fuchsia 控制器使用 FIDL 中间表示法 (FIDL IR) 在运行时生成其 FIDL 绑定。因此,您需要在 BUILD.gn
中为 fidlc
目标添加以下依赖项,以创建这些 FIDL 绑定:
"//src/developer/ffx/fidl:fuchsia.developer.ffx($fidl_toolchain)"
这还需要导入 $fidl_toolchain
变量:
import("//build/fidl/toolchain.gni")
如果您要编写测试,则需要添加主机测试数据(前提是基础架构测试也需要在测试运行程序上访问 IR),以便能够正常运行,例如:
"//src/developer/ffx/fidl:fuchsia.developer.ffx_host_test_data"
包含主机测试数据规则也将包含 FIDL IR,因此无需同时包含这两个依赖项。
添加 Python 导入块
添加完所有依赖项后,我们就可以在 Python 主文件中添加以下库:
from fuchsia_controller_py import Context, IsolateDir
import fidl.fuchsia_developer_ffx as ffx
import asyncio
以下部分将介绍此代码块中的各个库。
Context 和 IsolateDir
from fuchsia_controller_py import Context, IsolateDir
第一行包含一个 Context
对象,该对象提供了可供用户运行 ffx
命令的上下文。此外,您还可以使用此对象执行更多操作,因为它还提供了以下连接:
ffx
守护程序- Fuchsia 目标
IsolateDir
对象与 ffx
隔离相关,后者是指以某种方式运行 ffx
守护程序,使其所有元数据(例如配置值)都包含在特定目录下。隔离主要用于防止对 ffx
的状态造成污染,以及设置活跃程度较低的设备发现默认设置(这可能会在测试基础架构中运行 ffx
时导致出现问题)。
IsolateDir
对于通用命令是可选的,但如果您打算将程序用于测试,则必须填写。IsolateDir
对象会创建(并指向)一个目录,用于允许隔离的 ffx
守护程序实例运行。(如需详细了解 ffx
隔离,请参阅集成测试。)
需要在初始化期间将 IsolateDir
对象传递给 Context
对象。IsolateDir
对象也可以在 Context
对象之间共享。对 IsolateDir
对象进行垃圾回收后,ffx
守护程序也会关闭。
FIDL IR
import fidl.fuchsia_developer_ffx as ffx
第二行来自上一部分编写的 FIDL IR 代码。在 fidl.
之后写入的部分(例如 fuchsia_developer_ffx
)要求 fuchsia.developer.ffx
库存在 FIDL IR。任何 FIDL 导入行都是如此。若要导入 fidl.example_fuchsia_library
,需要为名为 example.fuchsia.library
的库生成 FIDL IR。使用 as
关键字可使此库易于使用。
此 fuchsia.developer.ffx
库包含 FIDL 绑定预期包含的所有结构,本教程的后面部分将对其进行介绍。
asyncio
import asyncio
从 FIDL IR 生成的对象使用异步绑定,这需要使用 asyncio
库。在本教程中,我们使用 echo.fidl
中定义的 echo 协议。
编写主要实现
除了 async_main
和 main
的样板之外,我们主要感兴趣的是 echo_func
定义:
async def echo_func():
isolate = IsolateDir()
config = {"sdk.root": "."}
ctx = Context(config=config, isolate_dir=isolate)
echo_proxy = ffx.Echo.Client(ctx.connect_daemon_protocol(ffx.Echo.MARKER))
echo_string = "foobar"
print(f"Sending string for echo: {echo_string}")
result = await echo_proxy.echo_string(value="foobar")
print(f"Received result: {result.response}")
async def async_main():
await echo_func()
def main():
asyncio.run(async_main())
if __name__ == "__main__":
main()
由于正在使用隔离,因此需要创建并传递给 Context
对象的 config
对象。当不再适合使用 ffx
的默认配置(默认情况下,ffx
知道在 Fuchsia 源代码检出设置中的什么位置找到 SDK)时,您要使用的任何配置值都必须提供给 Context
对象。
运行代码
我们必须先构建代码,然后才能运行代码。BUILD.gn
文件可能如下所示:
import("//build/python/python_binary.gni")
assert(is_host)
python_binary("example_echo") {
main_source = "main.py"
deps = [
"//src/developer/ffx:host",
"//src/developer/ffx/lib/fuchsia-controller:fidl_bindings",
"//src/developer/ffx/fidl:fuchsia.developer.ffx_compile_fidlc($fidl_toolchain)",
]
}
假设此 BUILD.gn
位于 src/developer/example_py_thing
目录中。然后,使用正确的 fx set
就位后,您可以使用主机目标构建此代码。如果您的主机是 x64
,构建命令可能如下所示:
fx build host_x64/obj/src/developer/example_py_thing/example_echo.pyz
构建完成后,您可以在 out
目录中找到代码(确切来说,默认为 out/default
)。而且,您可以直接从该目录运行 .pyz
文件。请务必使用 out/default
目录中的完整路径,以便 pyz
文件可以找到并打开相应的 .so
文件,例如:
$ cd $FUCHSIA_DIR/out/default
$ ./host_x64/obj/src/developer/example_py_thing/example_echo.pyz
Sending string for echo: foobar
Received result: foobar
$
与 Fuchsia 设备通信
如果该代码到目前为止已经构建并运行,我们就可以开始编写通过 FIDL 接口与 Fuchsia 设备通信的代码。大多数代码是相似的,但在此部分需要介绍一些细微的差异。
查找组件名称
如需绑定到 Fuchsia 组件,目前必须知道组件的名称。您可以使用 ffx
完成此操作。如需获取 build 信息提供程序的名称,例如:
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
您要使用的名称位于 Exposes
声明下:core/build-info
。
获取 build 信息
我们可以先从获取设备的 build 信息开始。
首先,我们需要添加 build 信息 FIDL 协议的依赖项:
"//sdk/fidl/fuchsia.buildinfo:fuchsia.buildinfo_compile_fidlc($fidl_toolchain)"
然后,我们需要编写代码,用于从 Fuchsia 设备获取代理。目前,这是通过连接到 build 信息 activity 名称实现的(不过,这项变更即将变更):
isolate = IsolateDir()
config = {"sdk.root": "."}
target = "foo-target-emu" # Replace with the target nodename.
ctx = Context(config=config, isolate_dir=isolate, target=target)
build_info_proxy = fuchsia_buildinfo.Provider.Client(
ctx.connect_device_proxy("/core/build-info", fuchsia_buildinfo.Provider.MARKER))
build_info = await build_info_proxy.get_build_info()
print(f"{target} build info: {build_info}")
如果运行上述代码,输出内容将如下所示:
foo-target-emu build info: ProviderGetBuildInfoResponse(build_info=BuildInfo(product_config='core', board_config='x64', version='2023-08-18T23:28:37+00:00', latest_commit_date='2023-08-18T23:28:37+00:00'))
如果要继续,则可以创建类似于 ffx target show
命令的内容:
results = await asyncio.gather(
build_info_proxy.get_build_info(),
board_proxy.get_info(),
device_proxy.get_info(),
...
)
由于对 FIDL 方法的每次调用都会返回一个协程,因此它们可以作为任务启动并并行等待,就像使用其他 FIDL 绑定时一样。
重新启动设备
重新启动设备的方法不止一种。一种重新启动设备的方法是连接到运行 fuchsia.hardware.power.statecontrol/Admin
协议的组件(可在 /bootstrap/shutdown_shim
下找到)。
如果使用此方法,协议应在方法执行过程中退出并显示 PEER_CLOSED
错误:
ch = self.device.ctx.connect_device_proxy(
"bootstrap/shutdown_shim", power_statecontrol.Admin.MARKER
)
admin = power_statecontrol.Admin.Client(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:
await coro
except ZxStatus as status:
if status.args[0] != ZxStatus.ZX_ERR_PEER_CLOSED:
raise status
不过,之后需要确定设备是否已恢复在线状态时,会面临一个颇具挑战性的问题。这通常通过尝试连接到协议(通常是 RemoteControl
协议)直到达到超时来实现。
另一种方法是连接到 ffx
守护程序的 Target
协议,可减少代码量:
ch = ctx.connect_target_proxy()
target_proxy = fuchsia_developer_ffx.Target.Client(ch)
await target_proxy.reboot(state=fuchsia_developer_ffx.TargetRebootState.PRODUCT)
运行组件
您可以使用 RemoteControl
协议启动组件,其中包含以下步骤:
连接到生命周期控制器:
ch = ctx.connect_to_remote_control_proxy() remote_control = fuchsia_developer_remotecontrol.RemoteControl.Client(ch) client, server = fuchsia_controller_py.Channel.create() await remote_control.root_lifecycle_controller(server=server.take()) lifecycle_ctrl = fuchsia_sys2.LifecycleController.Client(client)
尝试启动组件实例:
client, server = fuchsia_controller_py.Channel.create() await lifecycle_ctrl.start_instance("some_moniker", server=server.take()) binder = fuchsia_component.Binder.Client(client)
binder
对象可让用户知道该组件是否保持连接状态。不过,它没有方法。支持确定组件是否未绑定(使用 binder 协议)尚未实现。
获取快照
从紫红色设备获取快照涉及运行快照并绑定 File
协议进行读取:
client, server = Channel.create()
file = io.File.Client(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.DataProvider.Client(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)
asserts.assert_not_equal(response.response, None, extras=response)
response = response.response
if not response.data:
break
data.extend(response.data)
实现 FIDL 服务器
Fuchsia Controller 的一项重要任务(处理传递的绑定或测试复杂的客户端代码)是运行 FIDL 服务器。对于本教程中涵盖的所有 FIDL 协议,都有一个接受通道的客户端。为此,您需要使用 Server
类。
在本部分中,我们将返回 echo
示例,并实现一个 echo
服务器。您需要替换的函数派生自 FIDL 文件定义。因此,使用 ffx
协议的 echo
服务器如下所示:
class TestEchoer(ffx.Echo.Server):
def echo_string(self, request: ffx.EchoEchoStringRequest):
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.Echo.Server):
def echo_string(self, request: ffx.EchoEchoStringRequest):
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.Echo.Client(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.Example.Server):
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.Example.Server):
def check_file_exists(self, req: fe.ExampleCheckFileExistsRequests) -> fe.ExampleCheckFileExistsResponse | DomainError:
return DomainError(error=fe.SomeGenericError.THIS)
处理事件
事件处理脚本的编写方式与服务器类似,但它们派生自不同的基类 EventHandler
。事件在通道的客户端处理,因此必须传递客户端才能构建事件处理脚本。
我们先从以下 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.Example.EventHandler):
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
。
您可以使用一些现有的紫红色控制器测试代码测试此事件的示例。但首先,请确保 Fuchsia 控制器测试已添加到 Fuchsia build 设置中,例如:
fx set ... --with-host //src/developer/ffx/lib/fuchsia-controller: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.ExampleEvents.EventHandler):
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.ExampleEvents.Client(client_chan)
server = fct.ExampleEvents.Server(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
如需查看服务器测试的更多示例,请参阅此 server.py
文件。
试用 Python 解释器
如果您不确定如何构建某些类型,也不想构建和运行可执行文件,则可以使用 Python 解释器检查 FIDL 结构。
首先,请确保您已在 Python 中构建并可用于使用的必备 FIDL 库(在上一部分中进行了介绍),因为 Python 需要访问 FIDL IR 才能正常运行。
如需设置 Python 解释器,请运行以下命令(不过,这些命令取决于您的 Fuchsia build 目录,该目录默认为 $FUCHSIA_DIR/out/default
):
FUCHSIA_BUILD_DIR="$FUCHSIA_DIR/out/default" # Change depending on build dir.
export FIDL_IR_PATH="$FUCHSIA_BUILD_DIR/fidling/gen/ir_root"
__PYTHONPATH="$FUCHSIA_BUILD_DIR/host_x64:$FUCHSIA_DIR/src/developer/ffx/lib/fuchsia-controller/python"
if [ ! -z PYTHONPATH ]; then
__PYTHONPATH="$PYTHONPATH:$__PYTHONPATH"
fi
export PYTHONPATH="$__PYTHONPATH"
然后,您可以从任意位置启动 Python 解释器,该解释器还支持 Tab 键自动补全功能,以检查各种类型,例如:
$ python3
Python 3.11.8 (main, Feb 7 2024, 21:52:08) [GCC 13.2.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import fidl.fuchsia_hwinfo
>>> fidl.fuchsia_hwinfo.<TAB><TAB>
fidl.fuchsia_hwinfo.Architecture(
fidl.fuchsia_hwinfo.Board()
fidl.fuchsia_hwinfo.BoardGetInfoResponse(
fidl.fuchsia_hwinfo.BoardInfo(
fidl.fuchsia_hwinfo.Device()
fidl.fuchsia_hwinfo.DeviceGetInfoResponse(
fidl.fuchsia_hwinfo.DeviceInfo(
fidl.fuchsia_hwinfo.MAX_VALUE_SIZE
fidl.fuchsia_hwinfo.Product()
fidl.fuchsia_hwinfo.ProductGetInfoResponse(
fidl.fuchsia_hwinfo.ProductInfo(
fidl.fuchsia_hwinfo.fullname
您可以查看此模块导出的所有值。如果您想在 IPython
中试用异步函数,也可以执行与上文相同的环境设置并执行 IPython
。但首先,请确保您已安装 python3-ipython
:
sudo apt install python3-ipython
然后,您可以运行 IPython
。以下示例假定您运行名为 fuchsia-emulator
的模拟器并从 Fuchsia 默认 build 目录运行(否则,需要更改 "sdk.root"
):
Python 3.11.8 (main, Feb 7 2024, 21:52:08) [GCC 13.2.0]
Type 'copyright', 'credits' or 'license' for more information
IPython 8.20.0 -- An enhanced Interactive Python. Type '?' for help.
In [1]: from fuchsia_controller_py import Context
In [2]: import fidl.fuchsia_buildinfo
In [3]: ctx = Context(target="fuchsia-emulator", config={"sdk.root": "./sdk/exported/core"})
In [4]: hdl = ctx.connect_device_proxy("/core/build-info", fidl.fuchsia_buildinfo.Provider.MARKER)
In [5]: provider = fidl.fuchsia_buildinfo.Provider.Client(hdl)
In [6]: await provider.get_build_info()
Out[6]: ProviderGetBuildInfoResponse(build_info=BuildInfo(product_config='core', board_config='x64', version='2024-04-04T18:15:05+00:00', latest_commit_date='2024-04-04T18:15:05+00:00'))
...
如需详细了解如何使用 Fuchsia Controller 编写异步 Python 代码,请参阅此异步 Python 页面。