"""核心游戏引擎。 这一层是纯粹的"游戏规则"，不依赖终端、不依赖网络。 TUI 与 HTTP/gRPC 接口都调用同一组方法，因此三方共享完全一致的裁判逻辑。 线程模型 -------- 所有对状态的读写都在 ``self._lock`` 保护下进行，TUI 主线程、 自动推进线程、HTTP 工作线程可以安全并发调用。 错误约定 -------- 业务规则不满足时抛出 :class:`GameError`（带机器可读的 ``code``）， 调用方据此返回 4xx JSON 或界面提示。 """ from __future__ import annotations import contextlib import random import threading import time from collections.abc import Callable from typing import Any from . import config from .models import Animal, GameState, Plot, clamp class GameError(Exception): """业务规则错误。``code`` 供 API 调用方程序化判断。""" def __init__(self, code: str, message: str, status: int = 400) -> None: super().__init__(message) self.code = code self.message = message self.status = status def to_dict(self) -> dict[str, Any]: return {"ok": False, "error": self.code, "message": self.message} class FarmEngine: """农场游戏引擎。""" def __init__(self, tick_seconds: float = config.DEFAULT_TICK_SECONDS) -> None: self._lock = threading.RLock() self.state = GameState() self.tick_seconds = clamp( float(tick_seconds), config.MIN_TICK_SECONDS, config.MAX_TICK_SECONDS ) self._log_seq = 0 self.paused = False # 由 Scheduler 维护：下一次自动推进的绝对时间戳（无调度器时为 None） self.next_tick_at: float | None = None # 状态变化回调（TUI 用它触发重绘） self._listeners: list[Callable[[], None]] = [] # 开局先把今天与明天的天气掷好：否则第一天会顶着 DEFAULT_WEATHER 的 # 占位值开局，而"预报"字段又和它对不上。 self.state.weather = self._roll_weather() self.state.forecast = self._roll_weather(previous=self.state.weather) # ================================================================ 订阅 def subscribe(self, callback: Callable[[], None]) -> None: with self._lock: self._listeners.append(callback) def _notify(self) -> None: for callback in list(self._listeners): # 监听器异常不能影响游戏主流程 with contextlib.suppress(Exception): callback() # ================================================================ 日志 def add_log(self, text: str, kind: str = "info") -> None: with self._lock: self._log_seq += 1 self.state.log.append( { "seq": self._log_seq, "day": self.state.day, "kind": kind, # info / good / bad / warn / action "text": text, } ) if len(self.state.log) > config.MAX_LOG_ENTRIES: del self.state.log[: len(self.state.log) - config.MAX_LOG_ENTRIES] def recent_log(self, limit: int = 12, since: int = 0) -> list[dict[str, Any]]: with self._lock: entries = [e for e in self.state.log if e["seq"] > since] return entries[-limit:] if limit > 0 else entries # ================================================================ 查询 def snapshot(self) -> dict[str, Any]: """返回完整状态快照（供 TUI 渲染与 API 响应）。""" with self._lock: data = self.state.to_dict() data["tick_seconds"] = self.tick_seconds data["paused"] = self.paused data["seconds_to_tick"] = self.seconds_to_tick data["last_event"] = self.state.log[-1] if self.state.log else None return data @property def seconds_to_tick(self) -> float | None: if self.paused or self.next_tick_at is None: return None return round(max(0.0, self.next_tick_at - time.time()), 1) def brief_source(self) -> dict[str, Any]: """精简状态（动作回执、排行榜、`status`）要用的字段。 刻意**不走** :meth:`snapshot`：那些场景只要十几个标量，而完整快照会把 每块地、每只牲畜都序列化成 dict，再被调用方整批丢掉。实测每个请求因此 白花 ~60µs；排行榜更夸张——N 个牧场就是 N 次全量序列化。 这里的键是 ``farm.server.app.BRIEF_KEYS`` 的超集：加字段时两边一起改。 """ with self._lock: state = self.state # net_worth 是这里最贵的一项（要遍历地块与牲畜），算一次给两处用 worth = state.net_worth return { "day": state.day, "gold": state.gold, "level": state.level, "exp": state.exp, "exp_needed": state.exp_needed, "ap": state.ap, "max_ap": state.max_ap, "feed": state.feed, "plot_count": len(state.plots), "max_plots": state.max_plots, "next_plot_price": state.next_plot_price, "animal_count": len(state.animals), "max_animals": config.MAX_ANIMALS, "net_worth": round(worth, 2), "goal_index": config.goal_index(worth), "seconds_to_tick": self.seconds_to_tick, "paused": self.paused, "tick_seconds": self.tick_seconds, "weather": state.weather, "weather_name": config.weather_info(state.weather)["name"], "forecast": state.forecast, "forecast_name": config.weather_info(state.forecast)["name"], "storage_used": state.storage_used, "granary_capacity": state.granary_capacity, "granary_free": state.granary_free, # 存货是给 AI 的"卖什么"清单；只带非空项，通常只有几行 "inventory": dict(state.inventory), "building_count": sum(state.buildings.values()), # 待办清单是特意为 AI 准备的：直接读模型对象，不再过一遍序列化 "ripe_plots": [p.index + 1 for p in state.plots if p.is_ripe], "hungry_animals": [a.aid for a in state.animals if not a.fed], "collectable_animals": [a.aid for a in state.animals if a.pending > 0], } def market_price(self, key: str) -> float: with self._lock: return round(self.state.market.get(key, 0.0), 2) def catalog(self) -> dict[str, Any]: """返回所有可种植/可养殖对象及其在市场上的实时价格。""" with self._lock: crops = [] for key, info in config.CROPS.items(): price = self.market_price(key) gross = price days = int(info["grow_days"]) crops.append( { "key": key, "name": info["name"], "seed_cost": info["seed_cost"], "grow_days": days, "price": price, "exp": info["exp"], "level_req": info["level_req"], "locked": self._is_locked(info), "profit": round(gross - info["seed_cost"], 2), "profit_per_day": round((gross - info["seed_cost"]) / days, 2), } ) animals = [] for key, info in config.ANIMALS.items(): price = self.market_price(f"animal:{key}") feed_cost = info["feed_per_day"] * config.FEED_UNIT_PRICE * info["cycle"] animals.append( { "key": key, "name": info["name"], "cost": info["cost"], "product": info["product"], "price": price, "cycle": info["cycle"], "feed_per_day": info["feed_per_day"], "exp": info["exp"], "level_req": info["level_req"], "locked": self._is_locked(info), "profit_per_day": round((price - feed_cost) / max(1, info["cycle"]), 2), } ) goods = self._goods_catalog() return { "crops": crops, "animals": animals, "goods": goods, "buildings": self._building_catalog(), "weather": self._weather_catalog(), "granary": { "used": self.state.storage_used, "capacity": self.state.granary_capacity, }, "feed_price": config.FEED_UNIT_PRICE, "water_capacity": config.WATER_CAPACITY, "ap_cost": config.AP_COST, } # ---- catalog 的三个子表。拆开是因为它们的"变化频率"各不相同： # goods 随市价与存货变，buildings 随等级与已建数量变，weather 每天都在变。 def _goods_catalog(self) -> list[dict[str, Any]]: """所有可卖货物（作物 + 畜产品）的现价、基准价与仓库存量。 合成一张表是刻意的：``sell`` 的 key 空间就是这张表的 key 空间， 客户端不必自己拼 "animal:chicken" 这种前缀。 """ state = self.state goods: list[dict[str, Any]] = [] keys = list(config.CROPS) + [f"animal:{kind}" for kind in config.ANIMALS] for key in keys: price = self.market_price(key) base = config.base_price(key) stock = state.inventory.get(key, 0) goods.append( { "key": key, "name": config.goods_name(key), "kind": "animal" if key.startswith("animal:") else "crop", "price": price, "base_price": base, "ratio": round(price / base, 3) if base else 1.0, "stock": stock, "value": round(price * stock, 2), } ) return goods def _building_catalog(self) -> list[dict[str, Any]]: return [self._building_entry(self.state, key) for key in config.BUILDINGS] def _building_entry(self, state: GameState, key: str) -> dict[str, Any]: """一座建筑的完整状态：报价、已建、上限、解锁。 catalog 与 available_actions 都读它——否则"报价"和"建议"两处 一旦各写一遍，迟早就出现"建议建、真去建却说太贵"这种矛盾。 """ info = config.BUILDINGS[key] owned = state.building_count(key) maxed = owned >= info["max_level"] return { "key": key, "name": info["name"], "effect": info["effect"], "cost": info["cost"], "next_cost": -1 if maxed else config.build_price(key, owned), "level_req": info["level_req"], "locked": state.level < info["level_req"], "owned": owned, "max_level": info["max_level"], "maxed": maxed, } def _weather_catalog(self) -> dict[str, Any]: """今天的天气、明天的预报，以及这张天气表本身（给 AI 讲清规则用）。""" state = self.state return { "today": state.weather, "today_name": config.weather_info(state.weather)["name"], "forecast": state.forecast, "forecast_name": config.weather_info(state.forecast)["name"], "evaporation": self._evaporation(state, config.weather_info(state.weather)), "table": [ { "key": key, "name": info["name"], "desc": info["desc"], "evaporation": info["evaporation"], "water_gain": info["water_gain"], "damage": info["damage"], } for key, info in config.WEATHER.items() ], } def available_actions(self) -> list[dict[str, Any]]: """列出当前**合法**的动作，供 AI 直接决策使用。 每个动作包含：``action``（接口名）、``params``（参数模板）、 ``cost``（金币/行动点开销）、``reason``（为什么值得做）。 """ with self._lock: state = self.state actions: list[dict[str, Any]] = [] actions.extend(self._plot_actions(state)) actions.extend(self._animal_actions(state)) actions.extend(self._market_actions(state)) actions.extend(self._investment_actions(state)) actions.extend(self._build_actions(state)) return actions # ---- available_actions 的三个分区；拆开是为了各自能独立阅读/测试 def _plot_actions(self, state: GameState) -> list[dict[str, Any]]: """每块地当前能做的事：播种 / 收获 / 浇水。""" # 能种什么只取决于等级、金币和行动点，与**具体是哪块地**无关， # 所以整趟只筛一次，不要放进地块循环里做 O(地块 × 作物) 次判断。 plant_ap = config.AP_COST["plant"] plantable = [ (key, info) for key, info in config.CROPS.items() if not self._is_locked(info) and state.gold >= info["seed_cost"] and state.ap >= plant_ap ] actions: list[dict[str, Any]] = [] for plot in state.plots: position = plot.index + 1 if plot.crop is None: for key, info in plantable: actions.append( { "action": "plant", "params": {"plot": position, "crop": key}, "cost": {"gold": info["seed_cost"], "ap": plant_ap}, "reason": f"{position} 号地空闲，可种{info['name']}", } ) elif plot.is_ripe: actions.append( { "action": "harvest", "params": {"plot": position}, "cost": {"gold": 0, "ap": config.AP_COST["harvest"]}, "reason": f"{position} 号地的{config.CROPS[plot.crop]['name']}已成熟", } ) elif self._needs_water(plot) and state.ap >= config.AP_COST["water"]: reason = ( f"{position} 号地缺水，作物已停止生长" if plot.moisture <= 0 else f"{position} 号地水分只够撑到今天，不浇明天就会停长" ) actions.append( { "action": "water", "params": {"plot": position}, "cost": {"gold": 0, "ap": config.AP_COST["water"]}, "reason": reason, } ) return actions def _animal_actions(self, state: GameState) -> list[dict[str, Any]]: """每只牲畜当前能做的事：收取产出 / 喂食。""" actions: list[dict[str, Any]] = [] for animal in state.animals: if animal.pending > 0: actions.append( { "action": "collect", "params": {"animal_id": animal.aid}, "cost": {"gold": 0, "ap": config.AP_COST["collect"]}, "reason": f"{animal.name}有 {animal.pending} 份{animal.product}待收取", } ) elif not animal.fed: need = animal.info["feed_per_day"] cost = 0 if state.feed >= need else need * config.FEED_EMERGENCY_PRICE if state.gold >= cost and state.ap >= config.AP_COST["feed"]: actions.append( { "action": "feed", "params": {"animal_id": animal.aid}, "cost": {"gold": cost, "ap": config.AP_COST["feed"]}, "reason": f"{animal.name}(#{animal.aid})今天还没喂食", } ) return actions def _investment_actions(self, state: GameState) -> list[dict[str, Any]]: """与具体地块/牲畜无关的长期投入：开垦土地、补饲料、买牲畜。""" actions: list[dict[str, Any]] = [] price = state.next_plot_price if price > 0 and state.gold >= price: actions.append( { "action": "buy_plot", "params": {}, "cost": {"gold": price, "ap": config.AP_COST["buy_plot"]}, "reason": f"可花 {price} 金币开垦第 {len(state.plots) + 1} 块地", } ) bulk_cost = config.FEED_UNIT_PRICE * config.FEED_BULK if state.feed < 5 and state.gold >= bulk_cost: actions.append( { "action": "buy_feed", "params": {"quantity": config.FEED_BULK}, "cost": {"gold": bulk_cost, "ap": config.AP_COST["buy_feed"]}, "reason": "饲料库存偏低，建议补货", } ) for key, info in config.ANIMALS.items(): if self._is_locked(info): continue if len(state.animals) >= config.MAX_ANIMALS or state.gold < info["cost"]: continue actions.append( { "action": "buy_animal", "params": {"kind": key}, "cost": {"gold": info["cost"], "ap": config.AP_COST["buy_animal"]}, "reason": f"可购买{info['name']}（{info['cost']} 金币）", } ) return actions def _market_actions(self, state: GameState) -> list[dict[str, Any]]: """该不该卖货：谷仓快满了要腾地方，价格明显高于基准价则值得出手。 这是"择时卖出"在 AI 侧的落点：把"现价 vs 基准价"直接算好写进 reason， 机器人/LLM 不需要自己记住每种货的基准价，也就能真的做出囤或抛的判断。 """ actions: list[dict[str, Any]] = [] if not state.inventory: return actions used, capacity = state.storage_used, state.granary_capacity tight = used >= capacity * 0.8 for key, count in sorted(state.inventory.items()): if count <= 0: continue price = self.market_price(key) base = config.base_price(key) ratio = price / base if base else 1.0 reasons = [f"谷仓有 {count} 份{config.goods_name(key)}，现价 {price}（基准 {base}）"] if tight: reasons.append(f"谷仓已用 {used}/{capacity}，该腾地方了") if ratio >= 1.1: reasons.append(f"比基准价高 {(ratio - 1) * 100:.0f}%，适合出手") if len(reasons) == 1: # 价格没优势、仓也不挤 → 建议继续囤着。不推荐就别出声， # 否则候选动作里全是"卖了吧"，AI 反而挑不出重点。 continue actions.append( { "action": "sell", "params": {"key": key, "quantity": count}, "cost": {"gold": 0, "ap": config.AP_COST["sell"]}, "reason": "；".join(reasons), } ) return actions def _build_actions(self, state: GameState) -> list[dict[str, Any]]: """能造什么：等级够、没建满、且钱够。""" actions: list[dict[str, Any]] = [] for key in config.BUILDINGS: entry = self._building_entry(state, key) if entry["locked"] or entry["maxed"] or state.gold < entry["next_cost"]: continue actions.append( { "action": "build", "params": {"key": key}, "cost": {"gold": entry["next_cost"], "ap": config.AP_COST["build"]}, "reason": f"可造第 {entry['owned'] + 1} 座{entry['name']}：{entry['effect']}", } ) return actions # ================================================================ 内部工具 def _is_locked(self, info: dict[str, Any]) -> bool: """这件作物/牲畜在当前等级是否还没解锁。 "等级门槛"这一条规则原本在 catalog / available_actions / plant / buy_animal / _gain_exp 里各写了一遍，改文案或改判定就得追五处。 现在只在这里定义，别处一律调它。 """ return self.state.level < info["level_req"] def _require_level(self, info: dict[str, Any]) -> None: """等级不够就抛 locked。播种与买牲畜都要过这一关。""" if self._is_locked(info): raise GameError( "locked", f"{info['name']}需要 Lv.{info['level_req']}，当前 Lv.{self.state.level}", ) def _newly_unlocked(self, level: int) -> list[str]: """正好在 level 这一级解锁的作物与牲畜名字（先作物后牲畜）。""" return [ info["name"] for table in (config.CROPS, config.ANIMALS, config.BUILDINGS) for info in table.values() if info["level_req"] == level ] def _get_plot(self, plot_no: Any) -> Plot: """把 1-based 的地块编号转成 Plot，越界即报错。""" try: number = int(plot_no) except (TypeError, ValueError): # 底层 ValueError 的细节对调用者没意义，只保留友好提示 raise GameError("bad_param", f"地块编号必须是整数，收到 {plot_no!r}") from None if number < 1 or number > len(self.state.plots): raise GameError( "no_such_plot", f"地块编号 {number} 不存在（当前共 {len(self.state.plots)} 块地）", ) return self.state.plots[number - 1] def _get_animal(self, animal_id: Any) -> Animal: try: aid = int(animal_id) except (TypeError, ValueError): raise GameError("bad_param", f"动物编号必须是整数，收到 {animal_id!r}") from None for animal in self.state.animals: if animal.aid == aid: return animal raise GameError("no_such_animal", f"没有编号为 {aid} 的动物") def _spend_ap(self, action: str) -> None: cost = config.AP_COST.get(action, 0) if cost and self.state.ap < cost: raise GameError( "no_ap", f"行动点不足：{action} 需要 {cost} 点，当前只剩 {self.state.ap} 点。" f"等一天的时间自然走完就会恢复。", ) if cost: self.state.ap -= cost def _spend_gold(self, amount: int, what: str) -> None: amount = int(amount) if amount < 0: raise GameError("bad_param", "金额不能为负") if self.state.gold < amount: raise GameError( "not_enough_gold", f"金币不足：{what}需要 {amount}，你只有 {self.state.gold}", ) self.state.gold -= amount self.state.total_spent += amount def _earn(self, amount: float, reason: str) -> int: earned = max(0, round(amount)) self.state.gold += earned self.state.total_earned += earned return earned def _gain_exp(self, amount: int) -> list[str]: """增加经验并处理连续升级，返回升级提示文本。""" messages: list[str] = [] self.state.exp += int(amount) guard = 0 while self.state.exp >= self.state.exp_needed and guard < 50: guard += 1 self.state.exp -= self.state.exp_needed self.state.level += 1 bonus = config.level_up_bonus(self.state.level) self.state.gold += bonus unlocked = self._newly_unlocked(self.state.level) text = f"升级到 Lv.{self.state.level}！奖励 {bonus} 金币" if unlocked: text += f"，解锁：{'、'.join(unlocked)}" messages.append(text) self.add_log(text, "good") return messages def _roll_yield(self) -> int: """丰收回合判定：常规产出 1，暴击产出 2。""" return 2 if random.random() < config.LUCKY_CHANCE else 1 def _store(self, key: str, amount: int) -> None: """把货物放进谷仓；装不下就抛 granary_full，且**不改动任何状态**。 先判容量再落袋，所以收获撞上满仓时作物还好好地留在地里， 玩家卖掉一点东西就能重来，不会"收了个寂寞"。 """ amount = int(amount) if amount <= 0: return state = self.state free = state.granary_free if amount > free: raise GameError( "granary_full", f"谷仓装不下：还差 {amount - free} 格" f"（已用 {state.storage_used}/{state.granary_capacity}）。" "先 sell 卖掉点存货，或者 build 一座谷仓。", ) state.inventory[key] = state.inventory.get(key, 0) + amount def _needs_water(self, plot: Plot) -> bool: """这块地现在该不该浇水。 * 已经干透 → 今天就长不了，必须浇； * 水分刚好够今天蒸发、而明天又不下雨 → 明天会停长，提前浇能少浪费一天。 第二条用到了明天的预报——这就是"预报"这个字段存在的意义： 让"今天浇不浇"变成能提前算清的决策，而不是撞运气。 """ if plot.crop is None: return False if plot.moisture <= 0: return True forecast = config.weather_info(self.state.forecast) if forecast["water_gain"] > 0: # 明天有雨，今天浇水等于把水倒进雨里 return False return plot.moisture <= self._evaporation(self.state, forecast) # ================================================================ 动作 def plant(self, plot_no: Any, crop: Any) -> dict[str, Any]: with self._lock: if not isinstance(crop, str) or crop not in config.CROPS: raise GameError( "unknown_crop", f"未知作物 {crop!r}，可选：{'、'.join(config.CROPS)}", ) info = config.CROPS[crop] self._require_level(info) plot = self._get_plot(plot_no) if plot.crop is not None: raise GameError( "plot_busy", f"{plot.index + 1} 号地已经种了{config.CROPS[plot.crop]['name']}" ) self._spend_gold(info["seed_cost"], f"购买{info['name']}种子") self._spend_ap("plant") plot.crop = crop plot.growth = 0.0 plot.moisture = config.PLANT_INITIAL_MOISTURE plot.planted_day = self.state.day self.add_log(f"在 {plot.index + 1} 号地播下{info['name']}种子", "action") self._notify() return { "ok": True, "message": ( f"{plot.index + 1} 号地已种下{info['name']}，{info['grow_days']} 天后成熟" ), "plot": plot.to_dict(), } def water(self, plot_no: Any) -> dict[str, Any]: with self._lock: plot = self._get_plot(plot_no) if plot.crop is None: raise GameError("plot_empty", f"{plot.index + 1} 号地是空地，不需要浇水") if plot.moisture >= config.WATER_CAPACITY: raise GameError( "already_watered", f"{plot.index + 1} 号地水分已满（{plot.moisture}/{config.WATER_CAPACITY}）", ) self._spend_ap("water") plot.moisture = config.WATER_CAPACITY self.add_log(f"给 {plot.index + 1} 号地浇水（水分 {config.WATER_CAPACITY}）", "action") self._notify() return {"ok": True, "message": f"{plot.index + 1} 号地已浇水", "plot": plot.to_dict()} def harvest(self, plot_no: Any) -> dict[str, Any]: with self._lock: plot = self._get_plot(plot_no) if plot.crop is None: raise GameError("plot_empty", f"{plot.index + 1} 号地是空地") if not plot.is_ripe: raise GameError( "not_ripe", f"{config.CROPS[plot.crop]['name']}还差 {plot.days_left} 天成熟", ) crop_key = plot.crop info = config.CROPS[crop_key] self._spend_ap("harvest") quantity = self._roll_yield() # 先入库再清地：谷仓满了会在这里抛 granary_full，作物原地不动， # 玩家卖掉一点就能重来，不会"收了个寂寞"。 self._store(crop_key, quantity) price = self.market_price(crop_key) value = round(price * quantity, 2) plot.crop = None plot.growth = 0.0 plot.moisture = 0 self.state.total_harvested += quantity lucky = "（丰收！）" if quantity > 1 else "" self.add_log(f"收获 {quantity} 份{info['name']}{lucky}入库，现价 {price}", "good") levelups = self._gain_exp(info["exp"]) self._notify() return { "ok": True, "message": ( f"收获 {quantity} 份{info['name']}{lucky}入谷仓，现价 {price}" f"（约 {value} 金币，需 sell 才能换成钱）" ), "quantity": quantity, "price": price, "value": value, "exp": info["exp"], "level_ups": levelups, "storage_used": self.state.storage_used, } def buy_animal(self, kind: Any) -> dict[str, Any]: with self._lock: if not isinstance(kind, str) or kind not in config.ANIMALS: raise GameError( "unknown_animal", f"未知动物 {kind!r}，可选：{'、'.join(config.ANIMALS)}", ) info = config.ANIMALS[kind] self._require_level(info) if len(self.state.animals) >= config.MAX_ANIMALS: raise GameError("pasture_full", f"牧场已满（上限 {config.MAX_ANIMALS} 只）") self._spend_gold(info["cost"], f"购买{info['name']}") self._spend_ap("buy_animal") animal = Animal(kind) self.state.animals.append(animal) self.add_log(f"买了一只{info['name']}（#{animal.aid}），{info['cost']} 金币", "action") self._notify() return { "ok": True, "message": f"购入{info['name']} #{animal.aid}，记得每天喂食", "animal": animal.to_dict(), } def feed(self, animal_id: Any, quantity: Any = None) -> dict[str, Any]: with self._lock: animal = self._get_animal(animal_id) if animal.fed: raise GameError("already_fed", f"{animal.name} #{animal.aid} 今天已经吃过了") need = int(animal.info["feed_per_day"]) if quantity is None: quantity = need try: quantity = int(quantity) except (TypeError, ValueError): raise GameError("bad_param", "喂食数量必须是整数") from None if quantity <= 0: raise GameError("bad_param", "喂食数量必须大于 0") if quantity > need: quantity = need # 多喂无益，自动收窄到实际需求 if self.state.feed < quantity: shortfall = quantity - self.state.feed cost = shortfall * config.FEED_EMERGENCY_PRICE self._spend_gold(cost, f"临时采购 {shortfall} 份饲料") self.state.feed += shortfall self.add_log(f"饲料不足，临时高价补货 {shortfall} 份（-{cost} 金币）", "warn") self._spend_ap("feed") self.state.feed -= quantity animal.fed = True animal.starve = 0 self.add_log(f"喂了{animal.name} #{animal.aid} {quantity} 份饲料", "action") self._notify() return { "ok": True, "message": f"{animal.name} #{animal.aid} 吃饱了，剩余饲料 {self.state.feed}", "feed_left": self.state.feed, "animal": animal.to_dict(), } def collect(self, animal_id: Any) -> dict[str, Any]: with self._lock: animal = self._get_animal(animal_id) if animal.pending <= 0: raise GameError( "nothing_to_collect", f"{animal.name} #{animal.aid} 暂时没有可收取的{animal.product}" f"（还需 {animal.days_to_product} 天）", ) self._spend_ap("collect") quantity = animal.pending key = f"animal:{animal.kind}" # 与收获同理：先确认装得下，再清空待收，免得产出凭空消失 self._store(key, quantity) animal.pending = 0 price = self.market_price(key) value = round(price * quantity, 2) self.state.total_collected += quantity self.add_log( f"收取 {quantity} 份{animal.product}（{animal.name} #{animal.aid}）入库，" f"现价 {price}", "good", ) levelups = self._gain_exp(animal.info["exp"] * quantity) self._notify() return { "ok": True, "message": ( f"收取 {quantity} 份{animal.product}入谷仓，现价 {price}（约 {value} 金币）" ), "quantity": quantity, "price": price, "value": value, "level_ups": levelups, "storage_used": self.state.storage_used, } def buy_feed(self, quantity: Any = config.FEED_BULK) -> dict[str, Any]: with self._lock: try: quantity = int(quantity) except (TypeError, ValueError): raise GameError("bad_param", "饲料数量必须是整数") from None if quantity <= 0: raise GameError("bad_param", "饲料数量必须大于 0") if quantity > config.MAX_FEED_PURCHASE: raise GameError( "too_many", f"单次最多购买 {config.MAX_FEED_PURCHASE} 份饲料（请求 {quantity}）", ) cost = quantity * config.FEED_UNIT_PRICE self._spend_gold(cost, f"购买 {quantity} 份饲料") self.state.feed += quantity self.add_log(f"购买 {quantity} 份饲料（-{cost} 金币）", "action") self._notify() return { "ok": True, "message": f"购入 {quantity} 份饲料，库存 {self.state.feed}", "feed": self.state.feed, } def buy_plot(self) -> dict[str, Any]: with self._lock: price = self.state.next_plot_price if price < 0: raise GameError( "plot_limit", f"土地已达当前等级上限（{self.state.max_plots} 块），升级后可继续开垦", ) self._spend_gold(price, "开垦新土地") self._spend_ap("buy_plot") plot = Plot(len(self.state.plots)) self.state.plots.append(plot) self.state.plots_bought += 1 self.add_log(f"开垦了第 {plot.index + 1} 块地（-{price} 金币）", "action") self._notify() return { "ok": True, "message": f"新增 {plot.index + 1} 号地，共 {len(self.state.plots)} 块", "plot": plot.to_dict(), "next_plot_price": self.state.next_plot_price, } def sell(self, key: Any = None, quantity: Any = None) -> dict[str, Any]: """把谷仓里的货物按**当前市价**卖掉。 ``key`` 省略 = 能卖的都卖；``quantity`` 省略 = 该货物全卖。 这是存货变回金币的唯一出口，所以 ``granary_full`` 时它是唯一的解法—— ``available_actions`` 会在谷仓快满或价格见好时主动推荐它。 """ with self._lock: state = self.state if key is None: if quantity is not None: raise GameError("bad_param", "不指定 key 时不能只卖指定数量") if not state.inventory: raise GameError("nothing_to_sell", "谷仓还是空的，没东西可卖") return self._sell_everything(state) if not isinstance(key, str) or config.goods_info(key) is None: raise GameError( "unknown_goods", f"未知货物 {key!r}，可选：{'、'.join(sorted(state.market))}", ) name = config.goods_name(key) count = state.inventory.get(key, 0) if count <= 0: raise GameError("nothing_to_sell", f"谷仓里没有{name}") if quantity is None: quantity = count try: quantity = int(quantity) except (TypeError, ValueError): raise GameError("bad_param", "卖出数量必须是整数") from None if quantity <= 0: raise GameError("bad_param", "卖出数量必须大于 0") quantity = min(quantity, count) price = self.market_price(key) earned = self._earn(price * quantity, name) left = count - quantity if left > 0: state.inventory[key] = left else: del state.inventory[key] text = f"卖出 {quantity} 份{name}（单价 {price}），+{earned} 金币" self.add_log(text, "good") self._notify() return { "ok": True, "message": text, "key": key, "quantity": quantity, "price": price, "earned": earned, "gold": state.gold, "storage_used": state.storage_used, } def _sell_everything(self, state: GameState) -> dict[str, Any]: """一键清仓。带上明细，回执才能说清"到底卖了什么、各卖了多少"。""" sold: list[dict[str, Any]] = [] total = 0 for key in sorted(state.inventory): count = state.inventory[key] price = self.market_price(key) earned = round(price * count) total += earned sold.append( { "key": key, "name": config.goods_name(key), "quantity": count, "price": price, "earned": earned, } ) state.inventory.clear() self._earn(total, "卖出存货") names = "、".join(f"{item['name']}×{item['quantity']}" for item in sold) text = f"卖出 {names}，共 +{total} 金币" self.add_log(text, "good") self._notify() return { "ok": True, "message": text, "earned": total, "sold": sold, "gold": state.gold, "storage_used": state.storage_used, } def build(self, key: Any) -> dict[str, Any]: """造一座建筑。价格与解锁等级见 ``config.BUILDINGS``。""" with self._lock: if not isinstance(key, str) or key not in config.BUILDINGS: raise GameError( "unknown_building", f"未知建筑 {key!r}，可选：{'、'.join(config.BUILDINGS)}", ) entry = self._building_entry(self.state, key) if entry["locked"]: raise GameError( "locked", f"{entry['name']}需要 Lv.{entry['level_req']}，当前 Lv.{self.state.level}", ) if entry["maxed"]: raise GameError("build_limit", f"{entry['name']}已建满（{entry['max_level']} 座）") self._spend_gold(entry["next_cost"], f"建造{entry['name']}") self._spend_ap("build") self.state.buildings[key] = entry["owned"] + 1 built = self._building_entry(self.state, key) self.add_log( f"建成第 {built['owned']} 座{built['name']}（-{entry['next_cost']} 金币）", "good", ) self._notify() return { "ok": True, "message": ( f"{built['name']} 已建成（{built['owned']}/{built['max_level']}）：" f"{built['effect']}" ), "building": built, "gold": self.state.gold, } def pause(self, paused: bool | None = None) -> dict[str, Any]: with self._lock: self.paused = bool(paused) if paused is not None else not self.paused self.add_log("时间已暂停" if self.paused else "时间继续流动", "warn") self._notify() return {"ok": True, "paused": self.paused} def reset(self, keep_tick: bool = True) -> dict[str, Any]: with self._lock: tick = self.tick_seconds self.state = GameState() self._log_seq = 0 self.paused = False if keep_tick: self.tick_seconds = tick self.state.weather = self._roll_weather() self.state.forecast = self._roll_weather(previous=self.state.weather) self.add_log("新的一天，新的开始。欢迎来到 SimpleFarm！", "info") self._notify() return {"ok": True, "message": "农场已重置"} def save(self, path: str | None = None) -> dict[str, Any]: """把当前状态写入磁盘。 单机模式下由 TUI 直接调用；服务器模式里存档位置由牧场会话决定， 但保留同名方法是为了让界面代码不必区分"本地/远程"。 """ from .storage import save_state target = save_state(self.state, path) self.add_log(f"已存档到 {target}", "info") return {"ok": True, "message": f"已存档到 {target}", "path": target} # ================================================================ 天气与建筑 def _roll_weather(self, previous: str | None = None) -> str: """按权重掷一种天气；同一种不连着出现（最多重掷 WEATHER_REROLLS 次）。 权重已经表达了稀有度，连续性约束是额外补的一层手感——防的是 "一连五天干旱"这种既难玩、又不像真天气的序列。 """ keys = list(config.WEATHER) weights = [float(config.WEATHER[key]["weight"]) for key in keys] pick = keys[0] for _ in range(config.WEATHER_REROLLS): pick = random.choices(keys, weights=weights, k=1)[0] if pick != previous: return pick return pick def _resolve_weather(self, state: GameState, forced: str | None) -> dict[str, Any]: """定下今天的天气，并把预报翻到明天，返回今天天气的定义。 "预报"是刻意做成**提前一天可知**的：这样"明天要干旱，今晚先浇一遍" 才是一个算得清的决策，而不是撞运气。正常玩法走预报；``forced`` 只给测试与调试用（钉住"今天下雨"这类场景，否则断言会随掷骰抖动）。 """ if forced is not None and forced in config.WEATHER: today = forced elif state.forecast in config.WEATHER: today = state.forecast else: today = self._roll_weather() state.weather = today state.forecast = self._roll_weather(previous=today) return config.WEATHER[today] def _evaporation(self, state: GameState, info: dict[str, Any]) -> int: """当天实际蒸发量；温室让蒸发减半（向下取整）。 这是温室"省水"的那一半效果，另一半是免疫暴风雨（见 _storm_damage）。 """ evaporate = int(info["evaporation"]) if state.building_count("greenhouse"): evaporate //= 2 return evaporate def _run_buildings(self, state: GameState) -> list[str]: """结算建筑的每日自动动作。 顺序是一条流水线：浇水（水井/喷灌）保住作物 → 收割机把熟了的收进 谷仓 → 磨坊把陈粮磨成饲料 → 喂食器喂牲口 → 售货机趁高价清库存。 共同的设计底线：**只替代玩家本来要花行动点做的事，绝不自动花钱** ——喂食器缺饲料停摆、收割机遇满仓停摆，都宁可停也不硬来。 """ return ( self._wells(state) + self._sprinklers(state) + self._harvesters(state) + self._mill(state) + self._feeders(state) + self._vendor(state) ) def _sprinklers(self, state: GameState) -> list[str]: """喷灌系统：水井的全量版，一口气浇完所有干透的地（限 1 座）。""" if state.building_count("sprinkler") <= 0: return [] events: list[str] = [] for plot in state.plots: if plot.crop is None or plot.moisture > 0: continue plot.moisture = config.WATER_CAPACITY events.append(f"喷灌系统给 {plot.index + 1} 号地浇了水") return events def _harvesters(self, state: GameState) -> list[str]: """自动收割机：每天替玩家收熟地，入库逻辑与手动收获完全一致。 刻意**不给经验**——机器收的不算你的劳动，手动收获才涨等级， 这是自动化换手感时留的手动价值。谷仓满了就停摆（同喂食器哲学）， 作物留在地里，卖了就能恢复。 """ quota = state.building_count("harvester") * config.HARVESTER_PLOTS if quota <= 0: return [] events: list[str] = [] for plot in state.plots: if quota <= 0: break if plot.crop is None or not plot.is_ripe: continue crop_key = plot.crop quantity = self._roll_yield() try: self._store(crop_key, quantity) except GameError: events.append("谷仓满了，自动收割机停摆") break plot.crop = None plot.growth = 0.0 plot.moisture = 0 state.total_harvested += quantity quota -= 1 events.append( f"自动收割机把 {plot.index + 1} 号地的{config.CROPS[crop_key]['name']}收进了谷仓" ) return events def _mill(self, state: GameState) -> list[str]: """饲料磨坊：把 1 份库存作物磨成饲料，挑**最便宜的**下手。 排在喂食器前面：早上磨好，喂食器当天就能吃上。只磨谷仓里的存货， 绝不自动买粮——"自动系统悄悄花光金币"是本项目的红线。 """ if state.building_count("mill") <= 0 or not state.inventory: return [] key = min(state.inventory, key=config.base_price) state.inventory[key] -= 1 if state.inventory[key] <= 0: del state.inventory[key] state.feed += config.MILL_OUTPUT return [f"饲料磨坊把 1 份{config.goods_name(key)}磨成了 {config.MILL_OUTPUT} 份饲料"] def _vendor(self, state: GameState) -> list[str]: """自动售货机：市价高于基准价的库存自动出手，越贵越先卖。 卖出是赚钱不是花钱，所以它是唯一"主动碰金币"的自动化——但每天 有份数上限，且基准价以下的存货绝不碰，留着等高价。 """ quota = state.building_count("vendor") * config.VENDOR_QUOTA if quota <= 0 or not state.inventory: return [] offers = [] for key in sorted(state.inventory): base = config.base_price(key) if base <= 0: continue ratio = self.market_price(key) / base if ratio > 1.0: offers.append((ratio, key)) offers.sort(reverse=True) events: list[str] = [] for _, key in offers: if quota <= 0: break qty = min(state.inventory[key], quota) earned = self._earn(self.market_price(key) * qty, config.goods_name(key)) left = state.inventory[key] - qty if left > 0: state.inventory[key] = left else: del state.inventory[key] quota -= qty events.append(f"自动售货机卖出 {qty} 份{config.goods_name(key)}，+{earned} 金币") return events def _wells(self, state: GameState) -> list[str]: quota = state.building_count("well") if quota <= 0: return [] events: list[str] = [] for plot in state.plots: if quota <= 0: break if plot.crop is None or plot.moisture > 0: continue plot.moisture = config.WATER_CAPACITY quota -= 1 events.append(f"水井给 {plot.index + 1} 号地浇了水") return events def _feeders(self, state: GameState) -> list[str]: quota = state.building_count("feeder") if quota <= 0: return [] events: list[str] = [] for animal in state.animals: if quota <= 0: break if animal.fed: continue need = int(animal.info["feed_per_day"]) if state.feed < need: # 只报一次就停：饲料见底时每只都报一条会刷爆日志 events.append("饲料不够，自动喂食器停摆") break state.feed -= need animal.fed = True animal.starve = 0 quota -= 1 events.append(f"自动喂食器喂了{animal.name} #{animal.aid}") return events def _storm_damage(self, state: GameState, info: dict[str, Any]) -> list[str]: """暴风雨：每块**未成熟**的作物独立判定，被毁则地块直接清空。 成熟的作物不参与判定——"等你来收的东西被天灾吃掉"只会让人关掉游戏。 温室建成后整段失效。 """ damage = float(info["damage"]) if damage <= 0 or state.building_count("greenhouse"): return [] events: list[str] = [] for plot in state.plots: if plot.crop is None or plot.is_ripe or random.random() >= damage: continue name = config.CROPS[plot.crop]["name"] plot.crop = None plot.growth = 0.0 plot.moisture = 0 plot.planted_day = 0 text = f"暴风雨毁了 {plot.index + 1} 号地的{name}" events.append(text) self.add_log(text, "bad") return events # ================================================================ 时间推进 def advance_day(self, reason: str = "自动", weather: str | None = None) -> list[str]: """推进游戏内一天，返回本日发生的事件文本。 ``weather`` 是留给测试与调试的显式入口：不传就按"昨天的预报"结算 （正常玩法走这条），传了就强制使用该天气。 """ with self._lock: state = self.state info = self._resolve_weather(state, weather) state.day += 1 state.ap = state.max_ap events: list[str] = [] # ---- 建筑自动化。必须排在生长结算之前：水井救的正是"今天就断水"的 # 那块地，晚一步它今天就白等了。 events.extend(self._run_buildings(state)) # ---- 作物生长：水分消耗改由天气决定（0~2），浇水于是变成看天吃饭 evaporate = self._evaporation(state, info) gain = int(info["water_gain"]) for plot in state.plots: if plot.crop is None: continue if gain: # 雨天/暴风雨先补水再蒸发，等于免费替你浇了一遍 plot.moisture = min(config.WATER_CAPACITY, plot.moisture + gain) if plot.moisture <= 0: events.append(f"{plot.index + 1} 号地缺水，作物停止生长") continue plot.moisture = max(0, plot.moisture - evaporate) plot.growth = min(plot.grow_days, plot.growth + 1) if plot.is_ripe: events.append(f"{plot.index + 1} 号地的{config.CROPS[plot.crop]['name']}已成熟") # ---- 天灾：暴风雨只毁**还没成熟**的作物 events.extend(self._storm_damage(state, info)) # ---- 牲畜 dead: list[Animal] = [] for animal in state.animals: if animal.fed: animal.fed = False animal.starve = 0 animal.progress += 1 if animal.progress >= animal.cycle: animal.progress = 0 animal.pending += 1 else: animal.starve += 1 if animal.starve >= 3: dead.append(animal) elif animal.starve == 2: events.append(f"{animal.name} #{animal.aid} 已经两天没吃东西了") for animal in dead: state.animals.remove(animal) events.append(f"{animal.name} #{animal.aid} 因长期饥饿离你而去") self.add_log(f"{animal.name} #{animal.aid} 因长期饥饿离开农场", "bad") # ---- 市场波动 self._shuffle_market() # ---- 写入日志（最多 4 条，避免刷屏） for text in events[:4]: self.add_log(text, "warn") today = f"第 {state.day} 天开始（{reason}），今日{info['name']}" if events: self.add_log(f"{today}，行动点已恢复", "info") else: self.add_log(f"{today}，一切安好", "info") self._notify() return events def _shuffle_market(self) -> None: """每天让市场价格围绕基准价波动。""" vol = config.PRICE_VOLATILITY for key, info in config.CROPS.items(): base = float(info["base_price"]) factor = random.uniform(1 - vol, 1 + vol) # 轻微均值回归，避免价格长期漂移 current = self.state.market.get(key, base) blended = current * 0.35 + base * 0.65 self.state.market[key] = round(clamp(blended * factor, base * 0.55, base * 1.55), 2) for key, info in config.ANIMALS.items(): base = float(info["base_price"]) factor = random.uniform(1 - vol, 1 + vol) current = self.state.market.get(f"animal:{key}", base) blended = current * 0.35 + base * 0.65 self.state.market[f"animal:{key}"] = round( clamp(blended * factor, base * 0.55, base * 1.55), 2 ) def tick_decision( paused: bool, next_tick_at: float | None, tick_seconds: float, now: float ) -> tuple[bool, float]: """根据当前时刻决定"是否该推进一天"，并给出新的调度时间点。 这是 :class:`Scheduler`（单机）与 :class:`farm.server.session.FarmClock` （服务器）共用的唯一 tick 决策，两处不再各写一遍： * 暂停中 → 顺延，保证恢复后仍有完整的一个 tick； * 没有调度点（从未启动/被清空）→ 从现在起重新计时； * 到点 → 推进，并把下一次调度设为 now + tick_seconds； * 未到点 → 维持原调度时间。 返回 ``(should_advance, next_tick_at)``。 """ if paused or next_tick_at is None: return False, now + tick_seconds if now >= next_tick_at: return True, now + tick_seconds return False, next_tick_at class Scheduler: """自动推进游戏时间的后台线程（单机模式；服务器版见 farm.server.session.FarmClock）。""" def __init__(self, engine: FarmEngine, enabled: bool = True) -> None: self.engine = engine self.enabled = enabled self._stop = threading.Event() self._thread: threading.Thread | None = None def start(self) -> None: if not self.enabled or self._thread is not None: return self.engine.next_tick_at = time.time() + self.engine.tick_seconds self._thread = threading.Thread(target=self._run, name="farm-clock", daemon=True) self._thread.start() def stop(self) -> None: self._stop.set() self.engine.next_tick_at = None def _run(self) -> None: while not self._stop.wait(0.2): engine = self.engine advance, engine.next_tick_at = tick_decision( engine.paused, engine.next_tick_at, engine.tick_seconds, time.time() ) if advance: with contextlib.suppress(Exception): engine.advance_day(reason="时间流逝")