本教程逐步介绍了如何编写使用
Fuchsia 来源结账中的 Fuchsia 控制器 (fuchsia-controller
)
(fuchsia.git
) 设置。
Fuchsia Controller 由一组库组成,可让用户
并使用 FIDL 与设备进行交互。紫红色控制器
最初是为了测试而创建的但对于创建可编写脚本的
与 Fuchsia 设备上的 FIDL 接口交互的代码。例如,
用户可以使用 Fuchsia Controller 编写脚本,以执行简单的设备操作
而无需在 Rust 中编写 ffx
插件。
Fuchsia 控制器的两个主要部分是:
fuchsia-controller.so
文件(其中包括 ABI 的头文件)更高级别的语言绑定(基于
.so
文件构建) 使用 ABI)目前,Fuchsia Controller 的更高级别语言绑定是 。
使用 Fuchsia 控制器的最快方法是编写一个 Python 脚本,
使用 fuchsia-controller
代码。在 Fuchsia 源结账设置中,
你可以将 Python 二进制文件构建成一个 .pyz
文件,然后
从 out
目录执行(例如 $FUCHSIA_DIR/out/default
)。
如需编写您的第一个 Fuchsia Controller 脚本,请按以下步骤操作:
如果您遇到错误或者有任何问题或建议,请 提交 bug。
前提条件
本教程需要满足以下前提条件:
您需要使用 Fuchsia 源代码检出 (
fuchsia.git
) 开发 环境您需要运行 Fuchsia 设备。这可以是实体设备 或模拟器。
此设备必须连接到“
ffx
”并配有遥控器 。如果运行的是
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(${toolchain_variant.base})"
如果包含主机测试数据规则,则会同时包含 FIDL IR,因此无需 以包含这两个依赖项
添加 Python import 块
添加完所有依赖项后,我们就可以添加以下库 添加到 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
守护程序- 紫红色目标
IsolateDir
对象与 ffx
隔离相关,后者是指
运行 ffx
守护程序时,确保该守护程序的所有元数据(例如,config
值)包含在特定目录下。隔离主要是
旨在防止 ffx
的状态受到污染,并设置更少的
有效的设备发现默认值(这可能会导致在以下位置运行 ffx
时出现问题)
测试基础架构)。
对于通用命令,IsolateDir
是可选的;但如果您要
打算使用您的程序进行测试。IsolateDir
对象会创建(并且
指向)允许独立 ffx
守护程序实例运行的目录。
(如需详细了解 ffx
隔离,请参阅
集成测试。)
调用期间,需要将 IsolateDir
对象传递给 Context
对象
初始化。Context
之间也可以共享 IsolateDir
对象
对象的操作。清理 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
要求库的 FIDL IR
已生成名为“example.fuchsia.library
”的项目。使用 as
关键字
使此库易于使用。
此 fuchsia.developer.ffx
库包含预期的所有结构
(本教程稍后将对此进行介绍)。
asyncio
import asyncio
从 FIDL IR 生成的对象使用异步绑定,这需要
使用 asyncio
库。在本教程中,我们使用 echo 协议
在 echo.fidl
中定义。
编写主要实现
除了 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()
必须创建 config
对象并将其传递给 Context
对象
因为使用了隔离功能当它不再适用时
使用 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
,则 build
可能如下所示:
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
完成此操作。获取版本信息的名称
提供商,例如:
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 设备获取代理的代码。 目前,这通过连接到版本信息名称(尽管 (即将发生变更):
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 控制器的重要任务(处理传递的绑定)
或测试复杂的客户端代码)是运行 FIDL 服务器。所有角色
FIDL 协议,则有一个客户端接受
渠道。为此,您需要使用 Server
类。
在本部分中,我们返回到 echo
示例并实现一个 echo
服务器。
您需要替换的函数派生自 FIDL 文件定义。因此
echo
服务器(使用 ffx
协议)如下所示:
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 构建设置,例如:
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())
然后完成此 API 中的 Python 环境设置步骤, 下一部分。运行时,它会输出 以下输出并退出:
Got on-first event message: first message
Got on-second event
如需查看更多服务器测试示例,请参阅此 server.py
文件。
使用 Python 解释器进行实验
如果您不确定如何构建某些类型,并且不想构建和 运行可执行文件,则可以使用 Python 解释器检查 FIDL 结构。
首先,请确保您已构建且可用的必备 FIDL 库 以便在 Python 中使用(在上一部分中有所介绍),因为 Python 需要 FIDL IR 才能正常工作。
要设置 Python 解释器,请运行以下命令(不过,这些命令
取决于 Fuchsia 构建目录,该目录默认为
$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
中使用 async,您还可以进行与上文相同的环境设置,
执行 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 控制器编写异步 Python 代码, 请参阅异步 Python 页面。