本指南包含教學課程,說明如何編寫指令碼,以便遠端與 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 裝置相關資訊。
如果遇到錯誤,或有任何問題或建議,請回報問題。
必要條件
本教學課程需要以下項目:
- 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.gn
檔案,加入下列建構目標:
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
能力:
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 伺服器
執行 FIDL 伺服器是 Fuchsia Controller 的重要工作,可用於處理傳入的繫結或測試複雜的用戶端程式碼。在本節中,您將回到 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 建構設定,例如:
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
檔案。