How should I use the Dart bridge when developing Flutter Service extensions?

[yst-one]

When integrating FletService, there is no overridden didChangeDependencies method and no available context. How can I implement this line of code: _frames = FletBackend.of(context).openDataChannel();
below is the code example

import 'dart:typed_data';

import 'package:flet/flet.dart';
import 'package:flutter/material.dart';
import 'package:ultralytics_yolo/ultralytics_yolo.dart';

class YoloFletControl extends FletService {
  YOLO? yolo;
  late final DataChannel _frames;
  StreamSubscription<Uint8List>? _sub;
  YoloFletControl({
    required super.control,
  });

  @override
  void didChangeDependencies() {}

  @override
  void init() async {
    super.init();
    control.addInvokeMethodListener(_invokeMethod);

    if (_sub != null) return; // initialise lazily, once
    _frames = control.backend.c
    _sub = _frames.messages.listen(_onFrameFromPython);
    // Tell Python about the channel via a regular control event.
    control.triggerEvent("data_channel_open", {
      "channel_name": "frames",
      "channel_id": _frames.id,
    });

    final assetsSrc = control.backend.getAssetSource(control.getString("src")!);
    debugPrint("assetsSrc: $assetsSrc");
    yolo = YOLO(modelPath: assetsSrc.path);
    print('Loading model...');
    try {
      await yolo?.loadModel();
    } catch (e) {
      throw Exception("Error loading model: $e");
    }

    print('Model loaded successfully');
  }

  Future<Map<String, dynamic>> _detectObjects(Uint8List imageBytesList) async {
    final results = await yolo?.predict(imageBytesList);
    return results ?? {};
  }

  Future<dynamic> _invokeMethod(String name, dynamic args) async {
//     debugPrint("YoloFletControl.$name($args)");
    switch (name) {
      case "detectObjects":
//         debugPrint("YoloFletControl.$name($args)");
        final imageBytesList = args["image_bytes"];
//         debugPrint(
//             "imageBytesList: $imageBytesList, runtimeType: ${imageBytesList.runtimeType.toString()}");
        int startTime = DateTime.now().millisecondsSinceEpoch;
        final results = await _detectObjects(imageBytesList);
        final List<Map<String, dynamic>> results_list =
            results["boxes"]?.toList() ?? [];
        int endTime = DateTime.now().millisecondsSinceEpoch;
        debugPrint("推理耗时: ${endTime - startTime} 毫秒");
        return results_list;
      default:
        throw Exception("Method $name is not implemented");
    }
  }

  @override
  void dispose() {
    debugPrint("YoloFletControl(${control.id}).dispose()");
    yolo?.dispose();
    control.removeInvokeMethodListener(_invokeMethod);
    super.dispose();
  }
}

[yst-one]

@FeodorFitsner @ndonkoHenri

[chenjijun]

你用写的extend组件，可以导入其他项目中使用吗。为什么我写的一个xterm组件，在extend example中可以build后运行。 在其他项目中引入会显示控件不存在

[yst-one]

新建一个项目然后引入你写的插件试试

[ndonkoHenri]

@chenjijun An extension's Dart code is compiled into the Flutter client when you build the app that uses it. It isn't loaded at runtime. In your other project, check these:

1. Add the extension to that app's pyproject.toml. Put it in [project].dependencies. For a local, unpublished extension, also map it to its folder with [tool.flet.dev_packages], the way the template's example app does:

   [project]
   dependencies = ["flet", "flet-xterm"]
   
   [tool.flet.dev_packages]
   flet-xterm = "../flet-xterm"  # path relative to this pyproject.toml

   pip install -e and [tool.uv.sources] don't affect flet build.
2. Build that app with flet build <platform> -v (or flet debug <platform>). The log should contain Found Flutter extension at ....
3. Run the client you built. flet run without a build, flet run --web and the preview app all use the standard client, which shows Unknown control: Xterm.
4. Match the names. @ft.control("Xterm") in Python must be exactly the same as case "Xterm": in Dart.
5. Clear stale builds. After changing the extension, run flet clean or delete build/, then rebuild.

(中文) 扩展的 Dart 代码是在构建使用它的应用时编译进 Flutter 客户端的，不会在运行时加载。请检查：

1. 在该应用的 pyproject.toml 的 [project].dependencies 中加入扩展。本地扩展还要用 [tool.flet.dev_packages] 指向扩展目录（见上面的示例，路径相对于 pyproject.toml）。pip install -e 和 [tool.uv.sources] 对 flet build 无效。
2. 用 flet build <平台> -v（或 flet debug <平台>）构建，日志中应出现 Found Flutter extension at ...。
3. 运行构建出的客户端。未构建时的 flet run、flet run --web 以及预览 App 使用的都是标准客户端，会显示 Unknown control: Xterm。
4. Python 中 @ft.control("Xterm") 必须与 Dart 中的 case "Xterm": 完全一致。
5. 修改扩展后，先执行 flet clean 或删除 build/ 再重新构建。

[chenjijun]

@chenjijun An extension's Dart code is compiled into the Flutter client when you build the app that uses it. It isn't loaded at runtime. In your other project, check these:

1. Add the extension to that app's pyproject.toml. Put it in [project].dependencies. For a local, unpublished extension, also map it to its folder with [tool.flet.dev_packages], the way the template's example app does:

   [project]
   
   dependencies = ["flet", "flet-xterm"]
   
   
   
   [tool.flet.dev_packages]
   
   flet-xterm = "../flet-xterm"  # path relative to this pyproject.toml
   

   pip install -e and [tool.uv.sources] don't affect flet build.

2. Build that app with flet build <platform> -v (or flet debug <platform>). The log should contain Found Flutter extension at ....

3. Run the client you built. flet run without a build, flet run --web and the preview app all use the standard client, which shows Unknown control: Xterm.

4. Match the names. @ft.control("Xterm") in Python must be exactly the same as case "Xterm": in Dart.

5. Clear stale builds. After changing the extension, run flet clean or delete build/, then rebuild.

(中文) 扩展的 Dart 代码是在构建使用它的应用时编译进 Flutter 客户端的，不会在运行时加载。请检查：

1. 在该应用的 pyproject.toml 的 [project].dependencies 中加入扩展。本地扩展还要用 [tool.flet.dev_packages] 指向扩展目录（见上面的示例，路径相对于 pyproject.toml）。pip install -e 和 [tool.uv.sources] 对 flet build 无效。

2. 用 flet build <平台> -v（或 flet debug <平台>）构建，日志中应出现 Found Flutter extension at ...。

3. 运行构建出的客户端。未构建时的 flet run、flet run --web 以及预览 App 使用的都是标准客户端，会显示 Unknown control: Xterm。

4. Python 中 @ft.control("Xterm") 必须与 Dart 中的 case "Xterm": 完全一致。

5. 修改扩展后，先执行 flet clean 或删除 build/ 再重新构建。

Thanks for your reply. However, I feel the current import workflow for custom plugins is a bit cumbersome. My ideal workflow for custom plugin import would be:

1. Create a plugin project using the plugin example.

2. Debug the custom plugin and build the plugin.

3. Install the finished custom plugin into a new project, without requiring a separate build step. I want to debug directly with flet run.

4. Support all platforms that original Flutter plugins support.

Will the development workflow for custom Flutter plugins be updated in the future?

I’ve built a network device control center with Flet. It works with Flet Desktop and can be packaged via Docker. Inside the Docker container, I launch the app with flet run and a port parameter. This project has been really helpful for me. I want to add more custom Flutter plugins to extend this network device control center. Right now, my main missing feature is xterm integration for interactive SSH terminal access to network devices and support web and desktop.

[ndonkoHenri]

A FletService doesn't need a BuildContext. FletBackend.of(context) is only a Provider lookup, and every Control already has a reference to that same backend:

_frames = control.backend.openDataChannel();

This works from a service's init(), and from anywhere else you have a control.

Important

On the Python side, set on_data_channel_open before super().init().
In Dart, triggerEvent("data_channel_open", ...) is only sent if Python registered a handler. A Service is sent to Flutter from inside Service.init(), so a handler set after super().init() never reaches Flutter, and the event is silently dropped. Passing the handler to the constructor also works.
I am working on improving this.

Here's your service reworked. A few fixes beyond the channel line:

• init() stays synchronous. Flet doesn't await it, so void init() async hides model-load errors, and detectObjects can run before the model has loaded.
• dispose() cleans up. It cancels the subscription, closes the channel and disposes the model.
• No didChangeDependencies and no _sub guard. A service's init() only runs once.

Dart

import 'dart:async';
import 'dart:convert';
import 'dart:typed_data';

import 'package:flet/flet.dart';
import 'package:ultralytics_yolo/ultralytics_yolo.dart';

class YoloService extends FletService {
  YoloService({required super.control});

  DataChannel? _frames;
  StreamSubscription<Uint8List>? _framesSub;
  Future<YOLO>? _model;

  @override
  void init() {
    super.init(); // keep init() synchronous: Flet doesn't await it
    control.addInvokeMethodListener(_invokeMethod);

    // No BuildContext needed: control.backend is the same FletBackend
    // that FletBackend.of(context) returns.
    final frames = _frames = control.backend.openDataChannel();
    _framesSub = frames.messages.listen(_onFrame);
    control.triggerEvent("data_channel_open", {
      "channel_name": "frames",
      "channel_id": frames.id,
    });

    _model = _loadModel()..ignore(); // load errors surface in _detect()
  }

  @override
  void dispose() {
    control.removeInvokeMethodListener(_invokeMethod);
    _framesSub?.cancel();
    _frames?.close();
    unawaited(_model?.then((yolo) => yolo.dispose(), onError: (Object _) {}));
    super.dispose();
  }

  Future<YOLO> _loadModel() async {
    final src = control.getString("src");
    if (src == null) throw ArgumentError("src is not set");
    final yolo = YOLO(
      // absolute path of a file in the Flet app's assets dir
      modelPath: control.backend.getAssetSource(src).path,
      task: YOLOTask.detect,
    );
    if (!await yolo.loadModel()) throw StateError("Failed to load '$src'");
    return yolo;
  }

  Future<List<dynamic>> _detect(Uint8List image) async {
    final yolo = await _model!; // waits while the model is still loading
    final result = await yolo.predict(image);
    return result["boxes"] as List? ?? const [];
  }

  // One-off images: `await yolo.detect_objects(jpeg)` in Python.
  Future<dynamic> _invokeMethod(String name, dynamic args) async {
    switch (name) {
      case "detectObjects":
        return _detect(args["image_bytes"] as Uint8List);
      default:
        throw Exception("Method $name is not implemented");
    }
  }

  // Camera frames over the DataChannel.
  // Python -> Dart: [request id: 4 bytes][JPEG/PNG]; Dart -> Python: [same id][JSON]
  Future<void> _onFrame(Uint8List packet) async {
    if (packet.length < 4) return;
    final frame = Uint8List.fromList(packet); // own it: the buffer may be reused
    Object reply;
    try {
      reply = {"boxes": await _detect(Uint8List.sublistView(frame, 4))};
    } catch (e) {
      reply = {"error": "$e"};
    }
    _frames?.send(Uint8List.fromList(
        [...frame.take(4), ...utf8.encode(jsonEncode(reply))]));
  }
}

class Extension extends FletExtension {
  @override
  FletService? createService(Control control) {
    switch (control.type) {
      case "YoloService": // must match @ft.control("YoloService") in Python
        return YoloService(control: control);
      default:
        return null;
    }
  }
}

Python

import asyncio
import json
from typing import Any, Optional

import flet as ft


@ft.control("YoloService")  # must match `case "YoloService":` in Dart
class YoloService(ft.Service):
    src: Optional[str] = None
    on_data_channel_open: Optional[ft.EventHandler[ft.DataChannelOpenEvent]] = None

    def init(self):
        self._channel: Optional[ft.DataChannel] = None
        self._channel_ready = asyncio.Event()
        self._pending: dict[bytes, asyncio.Future] = {}
        self._next_id = 0
        # Set BEFORE super().init(): that's where the service is sent to Flutter,
        # and Dart only fires data_channel_open if a handler was set by then.
        if self.on_data_channel_open is None:
            self.on_data_channel_open = lambda e: None
        super().init()

    def before_event(self, e):
        # Grab the channel here, so it works even with a user-supplied handler.
        if isinstance(e, ft.DataChannelOpenEvent):
            self._loop = asyncio.get_running_loop()
            self._channel = self.get_data_channel(e.channel_id)
            self._channel.on_bytes(self._on_bytes)
            self._channel_ready.set()
        return super().before_event(e)

    def _on_bytes(self, payload: bytes):
        # can be called off the event-loop thread in built apps
        self._loop.call_soon_threadsafe(self._resolve, payload[:4], payload[4:])

    def _resolve(self, request_id: bytes, body: bytes):
        fut = self._pending.pop(request_id, None)
        if fut is not None and not fut.done():
            fut.set_result(json.loads(body))

    async def detect_frame(self, image: bytes, timeout: float = 30) -> list[dict[str, Any]]:
        """Camera frames over the DataChannel. Await each call before sending the next."""
        await asyncio.wait_for(self._channel_ready.wait(), timeout)
        self._next_id = (self._next_id + 1) & 0xFFFFFFFF
        request_id = self._next_id.to_bytes(4, "little")
        fut = self._pending[request_id] = asyncio.get_running_loop().create_future()
        self._channel.send(request_id + image)
        try:
            reply = await asyncio.wait_for(fut, timeout)
        finally:
            self._pending.pop(request_id, None)
        if "error" in reply:
            raise RuntimeError(reply["error"])
        return reply["boxes"]

    async def detect_objects(self, image: bytes, timeout: float = 30) -> list[dict[str, Any]]:
        """One-off images over the regular protocol."""
        return await self._invoke_method(
            "detectObjects", {"image_bytes": image}, timeout=timeout
        )

Usage:

yolo = YoloService(src="yolo11n.tflite")  # file in your assets dir
page.services.append(yolo)  # keeps a reference so the service isn't dropped

boxes = await yolo.detect_objects(jpeg_bytes)  # one-off image
boxes = await yolo.detect_frame(jpeg_bytes)    # camera stream, one frame at a time

A few notes:

• Camera streams: use detect_frame. Frames sent through invoke_method go through the regular control protocol. The DataChannel skips that and carries raw bytes.
• Platforms: ultralytics_yolo supports only Android and iOS. iOS needs a Core ML model (.mlpackage/.mlmodel), not .tflite.
• Testing: the Flet preview app (flet run --android/--ios) can't load custom Dart code, so use flet debug android|ios or flet build apk|ipa.
• What we tested: the code above passed unit tests with the YOLO plugin mocked. It hasn't run on a device yet, so please tell us how it goes.

Docs: https://flet.dev/docs/extend/user-extensions#dedicated-data-channels. The page only shows widgets right now; we'll add the service case.

[yst-one]

Thank you for your detailed explanation. Yes, we need to understand the difference between writing services and widgets. It would be ideal to have comprehensive documentation that illustrates the design philosophy of Flutter extensions.

[yst-one]

Android phone runs normally now. The 91KB data transfer to the Dart side is compressed to under 10 ms, which is a great improvement. There are two code warnings in VS Code. I hope to optimize the class to support code hints.