Metadata-Version: 2.4
Name: fastpilot
Version: 0.1.0
Summary: 让 Python 业务对象拥有统一的本地与远程控制、观察形状
Keywords: runtime,event-driven,ipython,remote-control
Author: 弥澄亮
Author-email: 弥澄亮 <t103ooooo@stu.mju.edu.cn>
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries
Requires-Dist: ipython>=9.0 ; extra == 'ipython'
Requires-Dist: httpx>=0.28,<1 ; extra == 'remote'
Requires-Dist: httpx-sse>=0.4.3,<0.5 ; extra == 'remote'
Requires-Dist: tenacity>=9.1,<10 ; extra == 'remote'
Requires-Dist: fastapi>=0.141,<1 ; extra == 'service'
Requires-Dist: uvicorn>=0.52,<1 ; extra == 'service'
Requires-Dist: pydantic>=2.13,<3 ; extra == 'service'
Requires-Dist: sse-starlette>=3.4,<4 ; extra == 'service'
Requires-Python: >=3.13
Provides-Extra: ipython
Provides-Extra: remote
Provides-Extra: service
Description-Content-Type: text/markdown

# FastPilot

FastPilot 让一个持续存在的 Python 业务对象，拥有统一的控制、观察与通信方式。

它适合这样的开发过程：你先在 IPython 里写出一个普通业务对象，然后逐步让
它能够接收类型化命令、保存需要继续判断的数据、发布领域事实、接受未来动作，
并在需要时由浏览器、远程 Python 或自动程序继续推动。

这段过程中，业务保持为你的 Python 代码。FastPilot 提供的是对象周围那条
可以继续观察和控制的公共因果链。

## 安装

FastPilot 需要 Python 3.13 或更高版本：

```bash
pip install fastpilot
```

核心包不强制安装网络、服务或 IPython 依赖。需要对应能力时，再安装可选依赖：

```bash
pip install "fastpilot[ipython]"
pip install "fastpilot[remote]"
pip install "fastpilot[service]"
```

```text
Command
    ↓
Participant
    ↓
Runtime
    ↓
Fact / Event / Snapshot / CommandReceipt
    ↓
IPython、远程 Python、浏览器、自动程序
```

## 先看一种日常写法

假设应用已经有一个普通的目录 SDK。现在我们想把“刷新目录”建模成一项可以
观察的业务动作，并保留后续判断需要的计数。

下面的代码可以按 IPython 的 `# %%` cell 逐段执行，也可以保存成普通 Python
脚本运行。每个 cell 都继续使用前面已经创建的对象。

### 第一格：声明控制输入和领域事实

```python
# %%
from dataclasses import dataclass
from typing import Literal

from fastpilot import (
    Command,
    Fact,
    ParticipantContext,
    command,
    emits,
    fact,
    handles,
    make_runtime,
)


@command("catalog.refresh")
@dataclass(frozen=True, slots=True)
class RefreshCatalog(Command):
    scope: Literal["recent", "all"]


@fact("catalog.refreshed")
@dataclass(frozen=True, slots=True)
class CatalogRefreshed(Fact):
    count: int
    scope: str
```

`RefreshCatalog` 表达“希望目录刷新”的控制输入；`CatalogRefreshed` 表达
“目录已经完成刷新”的领域事实。它们都是普通的不可变 Python dataclass，字段
就是业务需要传递的信息。

### 第二格：把已有业务对象接入 Runtime

```python
# %%
class CatalogClient:
    def fetch_catalog(self, scope: str) -> list[str]:
        records = ["book", "pen", "lamp"]
        return records if scope == "all" else records[:2]


class CatalogRefresher:
    initial_data = {"refresh_count": 0}

    def __init__(self, client: CatalogClient) -> None:
        self.client = client

    @handles
    @emits(CatalogRefreshed)
    def refresh(
        self,
        context: ParticipantContext,
        command: RefreshCatalog,
    ) -> None:
        records = self.client.fetch_catalog(command.scope)
        refresh_count = context.read_data("refresh_count", 0)
        assert isinstance(refresh_count, int)

        context.write_data("last_count", len(records))
        context.write_data("refresh_count", refresh_count + 1)
        context.emit(CatalogRefreshed(len(records), command.scope))


catalog = make_runtime(CatalogRefresher(CatalogClient()))
```

`CatalogRefresher` 是普通 Python 类。它通过构造参数持有已有的
`CatalogClient`，业务依赖由应用自己组织。

`@handles` 把 `RefreshCatalog` 交给 `refresh`；`@emits` 声明这个 handler
可以发布 `CatalogRefreshed`。`context` 是这次控制轮次使用的工作材料：
成功结束以后，写入的数据和发布的事实一起进入 Runtime 的新边界。

### 第三格：观察并推动一次控制轮次

```python
# %%
refreshed = catalog.subscribe(CatalogRefreshed)

catalog.start()
receipt = catalog.dispatch(RefreshCatalog("recent"))

print(receipt.outcome)
print(dict(receipt.snapshot.data))
print(refreshed.next(timeout=1).fact)
```

你会得到类似这样的结果：

```text
completed
{'refresh_count': 1, 'last_count': 2}
CatalogRefreshed(count=2, scope='recent')
```

这里有三种互相配合的观察方式：

* `receipt` 说明这一轮命令的结果；
* `receipt.snapshot` 保存这一轮结束时的完整边界；
* `subscribe()` 让另一个消费者沿自己的游标读取事实。

下一格可以继续使用 `catalog`、`receipt` 和 `refreshed`。这就是 FastPilot
最常见的工作感受：先让对象活在工作台里，再根据当前观察决定下一步。

## 这段代码解决了什么问题

普通函数通常适合这样的调用：

```python
result = refresh_catalog(scope="recent")
```

当业务开始需要持续存在时，调用者还会遇到另外一些问题：

* 上一次动作留下了什么数据？
* 这次命令实际产生了哪些事实？
* 另一个消费者怎样从自己的位置继续观察？
* 一个动作能不能稍后执行、重排或取消？
* 浏览器和远程 Python 怎样继续控制同一个对象？

FastPilot 把这些信息组织成 Runtime 的控制与观察边界。业务代码决定
自己的 Command、Fact 和数据；Runtime 负责把每次控制动作收束为可以继续
判断的 `CommandReceipt`、`Event` 和 `Snapshot`。

## 三个需要记住的角色

### Participant：你的业务规则

Participant 是普通 Python 对象。它拥有领域依赖和处理 Command 的方法：

```python
class CatalogRefresher:
    @handles
    @emits(CatalogRefreshed)
    def refresh(self, context: ParticipantContext, command: RefreshCatalog) -> None:
        ...
```

你可以从一个只有事件处理的 Participant 开始；当业务确实需要逻辑时间、
状态图或未来队列时，再让它声明对应能力。

### Runtime：持续存在的业务世界

```python
catalog = make_runtime(CatalogRefresher(CatalogClient()))

catalog.start()
catalog.dispatch(RefreshCatalog("recent"))
catalog.snapshot()
catalog.history()
```

Runtime 保存当前生命周期中的领域数据、事实历史、观察游标、队列、转换草稿、
可选逻辑时间和子 Runtime。它是同一条控制链的权威拥有者。

### Driver：根据观察选择下一步的人或程序

Driver 是一个角色，可以由下面这些对象承担：

* 持久 IPython 工作台；
* 远程 Python 程序；
* 浏览器控制台；
* 根据事件自动发送下一条命令的程序。

它们都可以沿着同一个 Runtime 的 `dispatch()`、`snapshot()`、`history()` 和
`subscribe()` 继续工作。

## 让未来动作先成为一个值

Command 可以立即执行，也可以登记为未来意图：

```python
# %%
from fastpilot import AtTime
from fastpilot.contrib import SetEnergy, make_robot

robot = make_robot()
robot.start()
entry = robot.enqueue(
    SetEnergy(50),
    trigger=AtTime(20),
)

robot.move_entry(entry.id, 0)
robot.cancel_entry(entry.id, "计划改变")
```

队列条目拥有稳定身份，因此未来动作可以被观察、重排、取消，并在历史中留下
自己的终态。事件驱动业务可以直接停留在前面的几格代码；需要时间能力的业务
再进入 Trigger 和队列。

## 需要状态图时，再加入转换站

状态图用于表达领域允许的状态边：

```python
# %%
from fastpilot import StateChart

statechart = StateChart.create(
    initial="IDLE",
    transitions={
        "IDLE": {"WALKING"},
        "WALKING": {"RESTING"},
        "RESTING": {"WALKING"},
    },
)
```

这个值可以由 Participant 作为自己的状态图提供：

```python
class RobotParticipant:
    statechart = statechart

    # 这里继续放置 Walk、Obstacle 等领域 Command 的 handler。
    ...
```

当领域 Command 提出状态变化时，Runtime 可以把下一条状态边停在一个可观察、
可编辑的 `TransitionDraft`：

```python
# %%
from fastpilot.contrib import Walk, make_robot

robot = make_robot()
robot.dispatch(Walk())

assert robot.pending is not None
robot.pending.context["operator_note"] = "沿北侧通道"
robot.commit()
robot.resume()
```

这里的日常表达很直接：先看提案，再补充工作材料，提交状态边，最后恢复后续
调度。需要这类人工介入的业务，可以在 `examples/03-robot` 中继续阅读。

## 需要网络时，让同一个对象打开入口

本地 Runtime 可以在当前进程中打开 HTTP/SSE 入口：

```python
# %%
service = catalog.serve(port=0)
print(service.url)
```

另一个 Python 进程使用同一个协议连接它：

```python
from fastpilot import connect

with connect(service.url) as remote:
    remote.dispatch(RefreshCatalog("all"))
```

远程调用发送同一种 `RefreshCatalog` Command，服务端由同一个 Runtime
完成解释、记录事件和生成快照。

如果应用本来就有 FastAPI，可以把控制面挂入已有 ASGI 应用：

```python
from typing import Literal

from fastapi import FastAPI
from pydantic import BaseModel
from fastpilot.service import mount_runtime, serve_app


class RefreshRequest(BaseModel):
    scope: Literal["recent", "all"]


app = FastAPI()
mount_runtime(app, catalog, prefix="/_pilot")


@app.post("/catalog/refresh")
def refresh(request: RefreshRequest) -> dict[str, object]:
    receipt = catalog.dispatch(RefreshCatalog(request.scope))
    return {
        "outcome": receipt.outcome,
        "sequence": receipt.snapshot.sequence,
    }


service = serve_app(app, port=0)
```

这段组合里：

* `RefreshRequest` 是 HTTP 边界模型；
* `RefreshCatalog` 是 Runtime 控制意图；
* `/catalog/refresh` 是应用自己的业务地址；
* `/_pilot/v1/*` 是同一个 Runtime 的通用控制与观察地址。

普通 FastAPI 路由与 FastPilot 控制面因此可以在一个应用里共同工作。

## 从哪里继续

这份 README 先覆盖安装和最小使用旅程。仓库中的 `examples/`、`docs/`、
`tests/` 和 `CONTRIBUTING.md` 属于开发维护材料，当前不随 PyPI 发布包提供；
它们会在整理成熟后再决定是否公开。
