透過指令碼與 Fuchsia 裝置進行遠端互動

本指南包含教學課程,說明如何編寫指令碼,以便遠端與 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 裝置相關資訊。

  1. 必要條件
  2. 建立建構目標
  3. 編寫指令碼
  4. 建構並執行指令碼

如果遇到錯誤,或有任何問題或建議,請回報問題

必要條件

本教學課程需要以下項目:

  • 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_remotecontrolfidl_fuchsia_buildinfo Python 模組包含 fuchsia.developer.ffxfuchsia.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.IdentifyHostfuchsia.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 通訊協定啟動元件,這項操作涉及下列步驟:

  1. 連線至生命週期控制器:

    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)
    
  2. 嘗試啟動元件的執行個體:

    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()

實作伺服器時,請注意以下幾點:

  • 方法定義可以是 syncasync
  • 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 檔案。