snowy-yang
a9bc8238ab
为什么：snapshot() 同时压在 SSE 每 250ms 推送和 full=1 动作回执两条
高频路径上，而它的成本大头是重复求值——profile 显示单次快照
1,128,001 次函数调用里近半是属性链重复计算同一份配置数据。
- Plot.to_dict：grow_days 原来经 4 个属性各求值一遍（本体 + days_left
+ is_ripe + stage），16 块地就是 64 次配置表查询。改为算一次局部
复用；stage 的实现体拆到 _stage(grow)，规则本体仍只有一份。
- Animal.to_dict：info 配置表原来查 4 次、cycle int() 两遍，改取一次。
- GameState.to_dict：weather_info 3 键 2 次查询；granary_free 原来把
granary_capacity 和 storage_used 整个重算，改算一轮复用。
行为零变化（规则本体全在原属性里，532 项测试全绿）。
基准（tools/bench.py --repeat 5，最坏形态 16 地 + 8 牲畜）：
场景 优化前 优化后
engine.snapshot() 50.1 µs 44.2 µs -12%
session.snapshot() 51.0 µs 41.9 µs -18%
build_state_message() 50.9 µs 43.3 µs -15%
engine.available_actions 52.7 µs 48.7 µs -8%
单次 snapshot 函数调用数 1,128,001 → 588,001（-48%），本次数字已记入
bench.py docstring 作长期参考。重复计数器全 OK（to_dict/snapshot/
net_worth 每请求仍各 1 次，无回归）。
438 lines
16 KiB
Python
438 lines
16 KiB
Python
"""游戏数据模型。
|
|
|
|
设计原则：模型只负责"持有数据 + 结构化转换"，所有规则判定放在 engine.py。
|
|
这样 HTTP 层和 TUI 层都能拿到同一份状态快照。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import itertools
|
|
from typing import Any
|
|
|
|
from . import config
|
|
|
|
|
|
def clamp(value: float, low: float, high: float) -> float:
|
|
"""把 value 限制在 [low, high] 区间内。"""
|
|
return max(low, min(high, value))
|
|
|
|
|
|
def format_duration(seconds: float | None) -> str:
|
|
"""把秒数写成人看的时长：``45s`` / ``39:58`` / ``1:05:00``。
|
|
|
|
一天默认 40 分钟（2400 秒），直接显示秒数是一串难读的大数，
|
|
所以 TUI 头部、``main.py state`` 这类地方都走这个函数。
|
|
"""
|
|
if seconds is None:
|
|
return "—"
|
|
total = int(max(0.0, seconds))
|
|
if total < 60:
|
|
return f"{total}s"
|
|
hours, rest = divmod(total, 3600)
|
|
minutes, secs = divmod(rest, 60)
|
|
if hours:
|
|
return f"{hours}:{minutes:02d}:{secs:02d}"
|
|
return f"{minutes}:{secs:02d}"
|
|
|
|
|
|
def next_goal_after(index: int) -> dict[str, Any] | None:
|
|
"""给定目标序号，返回"下一个目标"。已到最后一个则返回 None。"""
|
|
nxt = index + 1
|
|
if nxt >= len(config.GOALS):
|
|
return None
|
|
amount, text = config.GOALS[nxt]
|
|
return {"amount": amount, "text": text}
|
|
|
|
|
|
class Plot:
|
|
"""一块土地。"""
|
|
|
|
__slots__ = ("crop", "growth", "index", "moisture", "planted_day")
|
|
|
|
def __init__(self, index: int) -> None:
|
|
self.index = index # 0-based 下标，对外展示用 index+1
|
|
self.crop: str | None = None
|
|
self.growth: float = 0.0 # 已生长天数
|
|
self.moisture: int = 0 # 剩余可支撑天数
|
|
self.planted_day: int = 0
|
|
|
|
# ---------------------------------------------------------------- 查询
|
|
@property
|
|
def is_empty(self) -> bool:
|
|
return self.crop is None
|
|
|
|
@property
|
|
def grow_days(self) -> int:
|
|
if self.crop is None:
|
|
return 0
|
|
return int(config.CROPS[self.crop]["grow_days"])
|
|
|
|
@property
|
|
def is_ripe(self) -> bool:
|
|
return self.crop is not None and self.growth >= self.grow_days
|
|
|
|
@property
|
|
def days_left(self) -> int:
|
|
if self.crop is None:
|
|
return 0
|
|
return max(0, self.grow_days - int(self.growth))
|
|
|
|
@property
|
|
def stage(self) -> str:
|
|
"""人类可读的生长阶段。"""
|
|
return self._stage(self.grow_days)
|
|
|
|
def _stage(self, grow: int) -> str:
|
|
"""stage 的实现体。to_dict 已算好 grow_days，走这里避免重复求值。"""
|
|
if self.crop is None:
|
|
return "空地"
|
|
progress = self.growth / max(1, grow)
|
|
name = "已播种"
|
|
for threshold, label in config.GROWTH_STAGES:
|
|
if progress >= threshold:
|
|
name = label
|
|
return name
|
|
|
|
# ---------------------------------------------------------------- 序列化
|
|
def to_dict(self) -> dict[str, Any]:
|
|
# 热路径（每次快照对每块地调一次）：grow_days 原来经 4 个属性各求值
|
|
# 一遍（本处 + days_left + is_ripe + stage），这里算一次局部复用。
|
|
# 规则本体仍以同名属性为唯一来源，此处只是等价的直接计算。
|
|
info = config.CROPS.get(self.crop) if self.crop else None
|
|
grow = int(info["grow_days"]) if info else 0
|
|
return {
|
|
"index": self.index,
|
|
"position": self.index + 1,
|
|
"crop": self.crop,
|
|
"crop_name": info["name"] if info else None,
|
|
"growth": round(self.growth, 2),
|
|
"grow_days": grow,
|
|
"days_left": max(0, grow - int(self.growth)) if info else 0,
|
|
"is_ripe": info is not None and self.growth >= grow,
|
|
"is_empty": self.crop is None,
|
|
"moisture": self.moisture,
|
|
"water_capacity": config.WATER_CAPACITY,
|
|
"stage": self._stage(grow),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> Plot:
|
|
plot = cls(int(data.get("index", 0)))
|
|
crop = data.get("crop")
|
|
# 校验存档里的作物名仍然合法，避免旧存档导致 KeyError
|
|
plot.crop = crop if crop in config.CROPS else None
|
|
plot.growth = float(data.get("growth", 0.0))
|
|
plot.moisture = int(data.get("moisture", 0))
|
|
plot.planted_day = int(data.get("planted_day", 0))
|
|
return plot
|
|
|
|
|
|
class Animal:
|
|
"""一只牲畜。"""
|
|
|
|
_id_counter = itertools.count(1)
|
|
|
|
__slots__ = ("aid", "fed", "kind", "pending", "progress", "starve")
|
|
|
|
def __init__(self, kind: str, aid: int | None = None) -> None:
|
|
self.aid = aid if aid is not None else next(Animal._id_counter)
|
|
self.kind = kind
|
|
self.fed = False # 今天是否已喂食
|
|
self.progress = 0 # 距下次产出的累计天数
|
|
self.starve = 0 # 连续未进食天数
|
|
self.pending = 0 # 待收集的产出数量
|
|
|
|
# ---------------------------------------------------------------- 查询
|
|
@property
|
|
def info(self) -> dict[str, Any]:
|
|
return config.ANIMALS[self.kind]
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self.info["name"]
|
|
|
|
@property
|
|
def product(self) -> str:
|
|
return self.info["product"]
|
|
|
|
@property
|
|
def cycle(self) -> int:
|
|
return int(self.info["cycle"])
|
|
|
|
@property
|
|
def days_to_product(self) -> int:
|
|
return max(0, self.cycle - self.progress)
|
|
|
|
@property
|
|
def is_sick(self) -> bool:
|
|
"""连续 2 天没吃东西，进入"虚弱"状态。"""
|
|
return self.starve >= 2
|
|
|
|
# ---------------------------------------------------------------- 序列化
|
|
def to_dict(self) -> dict[str, Any]:
|
|
# 热路径：info 原来经 name/product/cycle/days_to_product 查 4 次配置表，
|
|
# 这里取一次局部复用；is_sick 与 days_to_product 的规则本体仍在属性里。
|
|
info = config.ANIMALS[self.kind]
|
|
cycle = int(info["cycle"])
|
|
return {
|
|
"id": self.aid,
|
|
"kind": self.kind,
|
|
"name": info["name"],
|
|
"product": info["product"],
|
|
"fed": self.fed,
|
|
"progress": self.progress,
|
|
"cycle": cycle,
|
|
"days_to_product": max(0, cycle - self.progress),
|
|
"pending": self.pending,
|
|
"starve": self.starve,
|
|
"is_sick": self.starve >= 2,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> Animal | None:
|
|
kind = data.get("kind")
|
|
if kind not in config.ANIMALS:
|
|
return None
|
|
animal = cls(kind, int(data.get("id", 0)) or None)
|
|
animal.fed = bool(data.get("fed", False))
|
|
animal.progress = int(data.get("progress", 0))
|
|
animal.starve = int(data.get("starve", 0))
|
|
animal.pending = int(data.get("pending", 0))
|
|
# 恢复 id 计数器，避免新动物与旧动物 id 冲突
|
|
Animal._id_counter = itertools.count(max(animal.aid + 1, next(Animal._id_counter)))
|
|
return animal
|
|
|
|
|
|
class GameState:
|
|
"""完整的游戏状态。所有字段都必须是可 JSON 序列化的。"""
|
|
|
|
def __init__(self) -> None:
|
|
self.day: int = 1
|
|
self.gold: int = config.START_GOLD
|
|
self.level: int = config.START_LEVEL
|
|
self.exp: int = 0
|
|
self.feed: int = config.START_FEED
|
|
self.plots: list[Plot] = [Plot(i) for i in range(config.BASE_PLOTS)]
|
|
self.animals: list[Animal] = []
|
|
self.log: list[dict[str, Any]] = []
|
|
|
|
# 市场价格（每天波动）：key -> 当前单价
|
|
self.market: dict[str, float] = {
|
|
key: float(info["base_price"]) for key, info in config.CROPS.items()
|
|
}
|
|
for key, info in config.ANIMALS.items():
|
|
self.market[f"animal:{key}"] = float(info["base_price"])
|
|
|
|
# 统计
|
|
self.total_harvested: int = 0
|
|
self.total_collected: int = 0
|
|
self.total_earned: int = 0
|
|
self.total_spent: int = 0
|
|
self.plots_bought: int = 0
|
|
|
|
# 谷仓：收获先进这里，卖不卖、什么时候卖由玩家决定。
|
|
# 键与 self.market 完全同构（作物用 "wheat"，畜产品用 "animal:chicken"），
|
|
# 这样定价、序列化、存档校验都只有一套规则。
|
|
self.inventory: dict[str, int] = {}
|
|
# 建筑：key -> 已建座数（0 表示还没造）
|
|
self.buildings: dict[str, int] = dict.fromkeys(config.BUILDINGS, 0)
|
|
# 天气：今天与明天。掷骰在 engine 里做，模型层只持有结果，
|
|
# 这样"存/读档"不需要重放随机序列就能复原当天的天气。
|
|
self.weather: str = config.DEFAULT_WEATHER
|
|
self.forecast: str = config.DEFAULT_WEATHER
|
|
|
|
# ---------------------------------------------------------------- 派生属性
|
|
@property
|
|
def max_plots(self) -> int:
|
|
"""当前等级允许持有的土地上限。"""
|
|
allowed = config.BASE_PLOTS + config.PLOTS_PER_LEVEL * (self.level - 1)
|
|
return int(clamp(allowed, config.BASE_PLOTS, config.MAX_PLOTS))
|
|
|
|
@property
|
|
def next_plot_price(self) -> int:
|
|
"""下一块土地的价格；已达上限时返回 -1。"""
|
|
if len(self.plots) >= self.max_plots:
|
|
return -1
|
|
return int(config.PLOT_BASE_PRICE * (config.PLOT_PRICE_GROWTH**self.plots_bought))
|
|
|
|
@property
|
|
def exp_needed(self) -> int:
|
|
return config.exp_to_next(self.level)
|
|
|
|
@property
|
|
def inventory_value(self) -> float:
|
|
return self.feed * config.FEED_UNIT_PRICE
|
|
|
|
@property
|
|
def livestock_value(self) -> float:
|
|
return sum(
|
|
a.pending * self.market.get(f"animal:{a.kind}", 0) + a.info["cost"] * 0.5
|
|
for a in self.animals
|
|
)
|
|
|
|
@property
|
|
def net_worth(self) -> float:
|
|
"""净资产：金币 + 已购地价 + 牲畜折价 + 饲料 + 地里未收的作物 + 谷仓存货。"""
|
|
land = sum(
|
|
config.PLOT_BASE_PRICE * (config.PLOT_PRICE_GROWTH**i) for i in range(self.plots_bought)
|
|
)
|
|
crops = 0.0
|
|
for plot in self.plots:
|
|
if plot.crop:
|
|
crops += self.market.get(plot.crop, 0) * (plot.growth / max(1, plot.grow_days))
|
|
stored = self.goods_value
|
|
return self.gold + land + self.livestock_value + self.inventory_value + crops + stored
|
|
|
|
@property
|
|
def goal_index(self) -> int:
|
|
return config.goal_index(self.net_worth)
|
|
|
|
@property
|
|
def next_goal(self) -> dict[str, Any] | None:
|
|
return next_goal_after(self.goal_index)
|
|
|
|
@property
|
|
def granary_capacity(self) -> int:
|
|
"""仓储上限：基础值 + 等级成长 + 谷仓建筑。"""
|
|
cap = (
|
|
config.GRANARY_BASE
|
|
+ config.GRANARY_PER_LEVEL * (self.level - 1)
|
|
+ config.GRANARY_PER_UPGRADE * self.building_count("granary")
|
|
)
|
|
return int(clamp(cap, config.GRANARY_BASE, config.MAX_GRANARY))
|
|
|
|
@property
|
|
def storage_used(self) -> int:
|
|
return sum(self.inventory.values())
|
|
|
|
@property
|
|
def granary_free(self) -> int:
|
|
return max(0, self.granary_capacity - self.storage_used)
|
|
|
|
@property
|
|
def goods_value(self) -> float:
|
|
"""谷仓存货按**当前市价**的估值（会随市场波动，这正是囤货的风险与收益）。"""
|
|
return sum(count * self.market.get(key, 0.0) for key, count in self.inventory.items())
|
|
|
|
def building_count(self, key: str) -> int:
|
|
return int(self.buildings.get(key, 0))
|
|
|
|
# ---------------------------------------------------------------- 序列化
|
|
def to_dict(self) -> dict[str, Any]:
|
|
# net_worth 要遍历地块与牲畜，是这里最贵的一项；下面三处都要用它，
|
|
# 所以只算一次再分发给 net_worth / goal_index / next_goal。
|
|
worth = self.net_worth
|
|
goal = config.goal_index(worth)
|
|
winfo = config.weather_info(self.weather)
|
|
finfo = config.weather_info(self.forecast)
|
|
# granary_free 原来会把 granary_capacity 和 storage_used 各重算一遍
|
|
cap = self.granary_capacity
|
|
used = self.storage_used
|
|
return {
|
|
"day": self.day,
|
|
"weather": self.weather,
|
|
"weather_name": winfo["name"],
|
|
"weather_desc": winfo["desc"],
|
|
"forecast": self.forecast,
|
|
"forecast_name": finfo["name"],
|
|
"gold": self.gold,
|
|
"level": self.level,
|
|
"exp": self.exp,
|
|
"exp_needed": self.exp_needed,
|
|
"feed": self.feed,
|
|
"inventory": dict(sorted(self.inventory.items())),
|
|
"storage_used": used,
|
|
"granary_capacity": cap,
|
|
"granary_free": max(0, cap - used),
|
|
"buildings": dict(self.buildings),
|
|
"plots": [p.to_dict() for p in self.plots],
|
|
"plot_count": len(self.plots),
|
|
"max_plots": self.max_plots,
|
|
"next_plot_price": self.next_plot_price,
|
|
"animals": [a.to_dict() for a in self.animals],
|
|
"animal_count": len(self.animals),
|
|
"max_animals": config.MAX_ANIMALS,
|
|
"market": {k: round(v, 2) for k, v in self.market.items()},
|
|
"net_worth": round(worth, 2),
|
|
"goal_index": goal,
|
|
"next_goal": next_goal_after(goal),
|
|
"stats": {
|
|
"harvested": self.total_harvested,
|
|
"collected": self.total_collected,
|
|
"earned": self.total_earned,
|
|
"spent": self.total_spent,
|
|
},
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> GameState:
|
|
state = cls()
|
|
state.day = max(1, int(data.get("day", 1)))
|
|
state.gold = max(0, int(data.get("gold", config.START_GOLD)))
|
|
state.level = max(1, int(data.get("level", 1)))
|
|
state.exp = max(0, int(data.get("exp", 0)))
|
|
state.feed = max(0, int(data.get("feed", 0)))
|
|
|
|
plots = [Plot.from_dict(p) for p in data.get("plots", []) if isinstance(p, dict)]
|
|
# 保证至少有初始土地，且不超过硬上限
|
|
count = int(clamp(len(plots) or config.BASE_PLOTS, config.BASE_PLOTS, config.MAX_PLOTS))
|
|
if len(plots) < count:
|
|
plots.extend(Plot(i) for i in range(len(plots), count))
|
|
state.plots = plots[: config.MAX_PLOTS]
|
|
for i, plot in enumerate(state.plots):
|
|
plot.index = i
|
|
|
|
animals: list[Animal] = []
|
|
for item in data.get("animals", [])[: config.MAX_ANIMALS]:
|
|
if isinstance(item, dict):
|
|
animal = Animal.from_dict(item)
|
|
if animal is not None:
|
|
animals.append(animal)
|
|
state.animals = animals
|
|
|
|
raw_buildings = data.get("buildings", {})
|
|
if isinstance(raw_buildings, dict):
|
|
for key, info in config.BUILDINGS.items():
|
|
try:
|
|
owned = int(raw_buildings.get(key, 0))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
state.buildings[key] = int(clamp(owned, 0, info["max_level"]))
|
|
|
|
weather = data.get("weather")
|
|
if isinstance(weather, str) and weather in config.WEATHER:
|
|
state.weather = weather
|
|
forecast = data.get("forecast")
|
|
if isinstance(forecast, str) and forecast in config.WEATHER:
|
|
state.forecast = forecast
|
|
|
|
market = data.get("market", {})
|
|
for key in list(state.market.keys()):
|
|
try:
|
|
value = float(market.get(key, state.market[key]))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
state.market[key] = clamp(value, 1.0, 100000.0)
|
|
|
|
# 存货键必须存在于 market：顺手挡住"手改存档塞进来的野键"，
|
|
# 否则它会一直躺在谷仓里、既卖不掉也算不进净资产。
|
|
raw_inventory = data.get("inventory", {})
|
|
if isinstance(raw_inventory, dict):
|
|
for key, count in raw_inventory.items():
|
|
if key not in state.market:
|
|
continue
|
|
try:
|
|
amount = int(count)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if amount > 0:
|
|
state.inventory[key] = amount
|
|
|
|
stats = data.get("stats", {})
|
|
state.total_harvested = max(0, int(stats.get("harvested", 0)))
|
|
state.total_collected = max(0, int(stats.get("collected", 0)))
|
|
state.total_earned = max(0, int(stats.get("earned", 0)))
|
|
state.total_spent = max(0, int(stats.get("spent", 0)))
|
|
state.plots_bought = max(0, len(state.plots) - config.BASE_PLOTS)
|
|
return state
|