plugin_info = {
    "name": "plugin_name",
    "api": 4,
    "version": (1, "0.1.0"),
    "expected_config": {"a": int},
    "initializer": function_that_do_something,
}

db_engine = Provide[Container.db_engine]

@inject
def function_that_needs_db(dbengine = Provide[Container.db_engine]): ...

@inject
def function_that_do_something(db_engine = Provide[Container.db_engine]): ...
    print(db_engine)

function_that_do_something()  # 在导入插件的过程中就运行

<dependency_injector.wiring.Provide object at 0xae9b0f50c152>

@inject
def after_import(db_engine: AsyncEngine = Provide[Container.db_engine]) -> None:
    print(db_engine)

plugin_info = {
    "name": "plugin_name",
    "api": 4,
    "version": (1, "0.1.0"),
    "expected_config": {"a": int},
    "initializer": after_import,
}

<sqlalchemy.ext.asyncio.engine.AsyncEngine object at 0x682669bc19c8>

class APlugin: ...

a_plugin = APlugin()  # 这很重要!

from pyfsd.plugin.interfaces import Plugin, PyFSDPlugin

class MyPlugin(Plugin, PyFSDPlugin):
    # 例如line_received_from_client事件
    def line_received_from_client(self, protocol: ClientProtocol, line: bytes) -> None:
        if line.startswith(b"#HI"):
            protocol.send_line(b"#HI:" + byte_line)
            # 在你确保你已经处理完这个事件后，可以抛出PreventEvent异常
            # 来停止传播此事件。
            raise PreventEvent
            # 注意! 目前你只能在line_received_from_client事件处理函数中抛出此异常


# 必须实例化，否则无法加载
plugin = MyPlugin()

from asyncio import get_event_loop
from typing import Optional

from aiohttp import ClientSession
from metar.Metar import Metar
from pyfsd.metar.fetch import MetarFetcher, MetarInfoDict
from pyfsd.plugin.interfaces import Plugin


class MetarFetcher(Plugin, MetarFetcher):
    metar_source = "example"  # Metar源名

    # 此处的config参数是配置文件的[pyfsd.metar]部分，下同
    async def fetch(self, config: dict, icao: str) -> Optional[Metar]:
        # 使用异步的aiohttp来下载Metar，不会使PyFSD整体阻塞
        # 详细用法请见aiohttp文档
        async with ClientSession() as session, session.get(url_of_metar) as resp:
            if resp.status != 200:  # 检查HTTP状态码是否是200
                return None
            metar_text = await resp.text("ascii", "ignore")
        if metar_text is None:
            return None
        else:
            return Metar(parser.metar_text, strict=False)

    async def fetch_all(self, config: dict) -> Optional[MetarInfoDict]:
        metar_text = {
            "ZSFZ": "ZSFZ 200300Z 17004MPS 9999 FEW020 33/27 Q1008 NOSIG",
            "ZBAA": "ZBAA 200400Z VRB02MPS CAVOK 34/21 Q1006 NOSIG",
        }  # 假设如此

        # 把解析部分单拎出来作为函数，方便在另外线程中运行
        def parser() -> MetarInfoDict:
            all_metar: MetarInfoDict = {}
            for icao, metar in metar_text.items():
                all_metar[icao] = Metar(metar, strict=False)
            return all_metar

        # fetch与fetch_all默认在主线程运行，所以fetch或fetch_all阻塞会导致PyFSD整个堵塞住
        # 这里使用asyncio的工具函数让parser函数在另一个线程运行，“异步化”
        return await get_event_loop().run_in_executor(None, parser)


# 必须实例化，否则无法加载
fetcher = MetarFetcher()

from asyncio import sleep
from time import time
from typing import Awaitable, Iterable

from pyfsd.plugin.interfaces import AwaitableMaker, Plugin


class AMaker(Plugin, AwaitableMaker):
    def __call__(self) -> Iterable[Awaitable]:
        async def func() -> None:  # 假设你那个阻塞的服务是这样的。。
            while True:
                print("Working", time())
                await sleep(1)

        yield func()  # 生成可等待对象
        # 接下来的代码会在PyFSD停止后运行，没需要留空也可以
        print("Clean up!")


# 必须实例化，否则无法加载
maker = AMaker()

2024-05-11 09:36:32 [info     ] PyFSD 0.0.1.2.dev0+86.gd72d5ce.dirty [pyfsd.main]
Working 1715391392.5279753
Working 1715391393.5289881
Working 1715391394.5294101
Working 1715391395.530576
2024-05-11 09:36:36 [info     ] Stopping                       [pyfsd.main]
Clean up!