snowy-yang
cf24cb43f1
动手前先修掉了两处「死机制」，新增的三个系统各自激活一个： * 天气：小麦 2 天成熟、初始水分 2、每天只掉 1 点，等于永远不需要浇水。 现在水分消耗由天气决定（晴 1 / 多云 0 / 干旱 2，雨天反补 3），而 `state.forecast` 把明天的天气提前告诉你——「今晚浇不浇」成了能算的决策。 5 种天气、权重合计 100、同种不连出；暴风雨会毁未熟作物，温室免疫。 * 谷仓：市场波动原本纯装饰（收获当场按市价变现，没有择时空间）。现在收获与 收取只进 `state.inventory`，靠新增的 `sell` 换金币，价格波动才真正参与决策。 仓满时 `_store()` 抛 `granary_full` 且不改状态，作物原样留在地里。 * 建筑：新增 `build`，买的是「以后不用天天手动做」——水井自动浇干地、 喂食器自动喂食（只吃库存饲料，绝不自动花钱）、温室、谷仓扩容。 每天在 `advance_day` 里先于生长执行，同类越造越贵（×1.6）。 存档升到 `SAVE_VERSION=2`，配 `MIN_SAVE_VERSION=1` 向后兼容，旧档缺的新字段 一律走默认值，不作废玩家手里的档。`advance_day(weather=...)` 是留给测试的 确定性入口。 顺带修两个真 bug： * `daemon.py` 三处 `makedirs(os.path.dirname(path))`：`--pid-file pid.txt` 这种只给文件名的用法会让 dirname 变成空串，启动阶段直接崩。抽出 `ensure_parent_dir()` 兜底。 * `RemoteFarm.snapshot()` 是本地缓存，非实时客户端轮询会一直读到旧数据， 看起来像服务器卡死。补 docstring 指明要自己调 `refresh()`。 质量门禁：ruff format/check 全绿、pyright 0 错误、503 项测试 OK（skip 35）、 bench 无回归（brief 19.5µs / snapshot 48.5µs / POST full=1 400µs）。 另起真实服务器（--tick 3 + 预置存档）跑了 38 项端到端验收，全过。
269 lines
10 KiB
Python
269 lines
10 KiB
Python
"""规则机器人玩家——"AI 也能玩"的最小可用示例。
|
||
|
||
它的意义不在于"内置了 AI"，而在于证明一件事：
|
||
**服务器把状态和合法动作都摊开了，外面是谁在决策根本不重要。**
|
||
这个机器人做决策的方式是"给候选动作打分"，而一个 LLM 做决策的方式是
|
||
"读同一份 ``/api/actions``，用语言模型选一个"。两者的接口调用序列一模一样。
|
||
|
||
所以：
|
||
|
||
* :meth:`RuleBot.choose` 里那段打分逻辑 = 可以整体替换成 LLM 的"策略函数"；
|
||
* :meth:`RuleBot.step` 的执行路径 = LLM 将来要走的那条路（读动作 → 执行 → 看回执）。
|
||
|
||
它只依赖 :mod:`farm.client`，因此既能在本机跑，也能从别处连上服务器跑——
|
||
换个 ``--url`` 就是一个独立的外挂玩家。
|
||
|
||
节奏完全由服务器时钟决定。服务器**没有**"手动推进一天"的接口，所以行动点
|
||
用完之后它只能等时间流动（或收工）：:meth:`RuleBot.run` 的 ``wait_seconds``
|
||
就是"最多等多久"，默认 0——一天 40 分钟，干等不划算。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import random
|
||
import threading
|
||
import time
|
||
from typing import Any
|
||
|
||
from .engine import GameError
|
||
from .models import format_duration
|
||
|
||
|
||
class RuleBot:
|
||
"""一个会种地、会养殖、会投资的规则机器人。"""
|
||
|
||
def __init__(
|
||
self, client, verbose: bool = True, pause: float = 0.25, name: str = "bot"
|
||
) -> None:
|
||
self.client = client # 任何与 RemoteFarm 接口一致的对象
|
||
self.verbose = verbose
|
||
self.pause = pause
|
||
self.name = name
|
||
self.done: list[str] = []
|
||
self.failures = 0
|
||
self.actions_executed = 0
|
||
self._stop = threading.Event()
|
||
|
||
# ------------------------------------------------------------ 输出
|
||
def say(self, text: str) -> None:
|
||
if self.verbose:
|
||
print(f"[{time.strftime('%H:%M:%S')}] {text}", flush=True)
|
||
|
||
def stop(self) -> None:
|
||
self._stop.set()
|
||
|
||
@property
|
||
def stopped(self) -> bool:
|
||
return self._stop.is_set()
|
||
|
||
# ------------------------------------------------------------ 打分
|
||
def score(
|
||
self, action: dict[str, Any], state: dict[str, Any], catalog: dict[str, Any]
|
||
) -> float:
|
||
"""给一个候选动作打分。分数越高越先执行。
|
||
|
||
这张表就是整个"策略"：先做免费的，再喂饱牲畜，再补水，再扩张。
|
||
换成 LLM 时，把这里换成一次模型调用即可，其余代码不用动。
|
||
"""
|
||
name = action.get("action")
|
||
params = action.get("params") or {}
|
||
gold = state.get("gold", 0)
|
||
feed = state.get("feed", 0)
|
||
ap = state.get("ap", 0)
|
||
|
||
if name == "harvest":
|
||
return 100.0
|
||
if name == "collect":
|
||
return 96.0
|
||
if name == "feed":
|
||
return 92.0
|
||
if name == "water":
|
||
return 84.0
|
||
if name == "plant":
|
||
# 优先种"每天净收益"最高的作物
|
||
crop = params.get("crop")
|
||
entry = next((c for c in catalog.get("crops", []) if c["key"] == crop), None)
|
||
per_day = entry["profit_per_day"] if entry else 0.0
|
||
return 60.0 + max(0.0, per_day)
|
||
if name == "buy_feed":
|
||
# 饲料快见底才补，且别把钱花光
|
||
if feed <= 2 and gold >= 200:
|
||
return 40.0
|
||
return 5.0
|
||
if name == "buy_plot":
|
||
price = state.get("next_plot_price") or 0
|
||
# 地太贵就别买，留下周转资金
|
||
if price > 0 and gold >= price * 3:
|
||
return 34.0
|
||
return 4.0
|
||
if name == "buy_animal":
|
||
info = next(
|
||
(a for a in catalog.get("animals", []) if a["key"] == params.get("kind")), None
|
||
)
|
||
cost = info["cost"] if info else 0
|
||
if cost and gold >= cost * 3 and ap >= 1:
|
||
return 30.0 + (info.get("profit_per_day", 0) if info else 0)
|
||
return 3.0
|
||
if name == "sell":
|
||
# 卖货是唯一的现金流入口（收获不再直接给钱），所以优先级跟着"缺钱"
|
||
# 和"仓快满"走：仓满了不腾地方，下一茬根本收不进来。
|
||
used = state.get("storage_used", 0) or 0
|
||
capacity = state.get("granary_capacity", 1) or 1
|
||
if gold < 150:
|
||
return 90.0
|
||
if used >= capacity * 0.8:
|
||
return 88.0
|
||
return 30.0
|
||
if name == "build":
|
||
key = params.get("key")
|
||
used = state.get("storage_used", 0) or 0
|
||
capacity = state.get("granary_capacity", 1) or 1
|
||
if key == "granary" and used >= capacity * 0.7:
|
||
return 46.0
|
||
if key in ("well", "feeder"):
|
||
# 这两个买的是"以后不用天天手动做"，放置玩法里稳赚
|
||
return 38.0
|
||
return 22.0
|
||
return 1.0
|
||
|
||
def choose(self) -> dict[str, Any] | None:
|
||
"""在服务器给出的合法动作里挑一个最优的。"""
|
||
actions = self.client.available_actions()
|
||
if not actions:
|
||
return None
|
||
state = self.client.snapshot()
|
||
catalog = self.client.catalog()
|
||
best = max(actions, key=lambda item: self.score(item, state, catalog))
|
||
if self.score(best, state, catalog) <= 1.0:
|
||
return None
|
||
return best
|
||
|
||
# ------------------------------------------------------------ 单步
|
||
def step(self) -> str | None:
|
||
"""执行一个动作，返回动作名；没得做或做不了返回 None。"""
|
||
action = self.choose()
|
||
if action is None:
|
||
return None
|
||
name = action["action"]
|
||
params = action.get("params") or {}
|
||
try:
|
||
result = self.client._action(name, params) # 走通用动作入口
|
||
except GameError as exc:
|
||
self.failures += 1
|
||
self.say(f"✗ {name} 失败（{exc.code}）：{exc.message}")
|
||
return None
|
||
self.actions_executed += 1
|
||
self.done.append(name)
|
||
self.say(f"✓ {name} {params} → {result.get('message', '')}")
|
||
if self.pause:
|
||
time.sleep(self.pause)
|
||
return name
|
||
|
||
def wait_for_next_day(self, timeout: float) -> bool:
|
||
"""没得可做时等时间自己流动到下一天。
|
||
|
||
返回 True 表示等到了新的一天；``timeout`` 为 0 表示不等待、直接收工
|
||
（机器人本来就该"把能做的做完就走"，不像真人会挂机守着）。
|
||
"""
|
||
state = self.client.snapshot()
|
||
if state.get("paused"):
|
||
self.say("— 时间被暂停了，再等也不会有下一天，先收工。")
|
||
return False
|
||
if timeout <= 0:
|
||
left = format_duration(state.get("seconds_to_tick"))
|
||
self.say(f"— 暂时没事可做，距离下一天还有 {left}，先收工。")
|
||
return False
|
||
|
||
start_day = state.get("day")
|
||
self.say(f"— 等时间流动（最多 {timeout:.0f} 秒）…")
|
||
deadline = time.time() + timeout
|
||
while not self._stop.is_set() and time.time() < deadline:
|
||
time.sleep(0.25)
|
||
day = self.client.snapshot().get("day")
|
||
if day != start_day:
|
||
self.say(f"— 天亮了：第 {day} 天")
|
||
return True
|
||
self.say("— 没等到下一天，先收工。")
|
||
return False
|
||
|
||
# ------------------------------------------------------------ 主循环
|
||
def run(
|
||
self,
|
||
rounds: int = 0,
|
||
max_minutes: float = 0.0,
|
||
idle_limit: int = 8,
|
||
wait_seconds: float = 0.0,
|
||
) -> dict[str, Any]:
|
||
"""跑机器人。
|
||
|
||
``rounds`` 为 0 表示不限轮数；``max_minutes`` 为 0 表示不限时间；
|
||
``wait_seconds`` 是"没事可做时最多等多久"（等下一天），0 表示直接收工。
|
||
"""
|
||
deadline = (time.time() + max_minutes * 60) if max_minutes > 0 else None
|
||
start_day = self.client.snapshot().get("day", 1)
|
||
idle = 0
|
||
round_no = 0
|
||
self.say(f"机器人 {self.name} 上线（第 {start_day} 天，{self.client.transport_label()}）")
|
||
|
||
while not self._stop.is_set():
|
||
round_no += 1
|
||
if rounds and round_no > rounds:
|
||
break
|
||
if deadline and time.time() > deadline:
|
||
self.say("到时间了，收工。")
|
||
break
|
||
|
||
state = self.client.snapshot()
|
||
if state.get("ap", 0) <= 0:
|
||
# 行动点用完 ≠ 没事可做：sell / build / harvest / collect / buy_*
|
||
# 都是 0 消耗，而 available_actions 本身就按行动点筛过一遍了，
|
||
# 所以照常挑一次最优的；真的没得选才等下一天。
|
||
if self.step() is None and not self.wait_for_next_day(wait_seconds):
|
||
break
|
||
idle = 0
|
||
continue
|
||
|
||
did = self.step()
|
||
if did is None:
|
||
idle += 1
|
||
if idle >= idle_limit:
|
||
if not self.wait_for_next_day(wait_seconds):
|
||
break
|
||
idle = 0
|
||
else:
|
||
time.sleep(0.4)
|
||
else:
|
||
idle = 0
|
||
|
||
end_day = self.client.snapshot().get("day", start_day)
|
||
summary = {
|
||
"name": self.name,
|
||
"days": end_day - start_day,
|
||
"actions": self.actions_executed,
|
||
"failures": self.failures,
|
||
"net_worth": self.client.snapshot().get("net_worth"),
|
||
"gold": self.client.snapshot().get("gold"),
|
||
}
|
||
self.say(
|
||
f"机器人 {self.name} 下线：跑了 {summary['days']} 天、"
|
||
f"执行 {summary['actions']} 个动作、净资产 {summary['net_worth']}"
|
||
)
|
||
return summary
|
||
|
||
|
||
def random_bot(client, **kwargs) -> RuleBot:
|
||
"""随机策略机器人：用来做对照实验（证明"有策略"确实比乱按强）。"""
|
||
bot = RuleBot(client, **kwargs)
|
||
original = bot.choose
|
||
|
||
def choose() -> dict[str, Any] | None:
|
||
actions = client.available_actions()
|
||
if not actions:
|
||
return None
|
||
return random.choice(actions)
|
||
|
||
bot.choose = choose # type: ignore[assignment]
|
||
bot.name = kwargs.get("name", "random")
|
||
del original
|
||
return bot
|