452 lines
18 KiB
GDScript
452 lines
18 KiB
GDScript
## VERTICAL SLICE - NOT FOR PRODUCTION
|
||
## 确定性棋盘生成器：生成-验证循环出图，不达标走接受最优/最不坏诚实降级（垂直切片验证：每张图可推理可归因）。
|
||
## 只消费传入的单条 rng 流（重试共用，attempt 边界非 scope 事件）——同流同参逐位复现。
|
||
## generate 的 db 参数鸭子型（不标类型）——冒烟测量面注入桩表依赖此形态，勿收紧签名。
|
||
class_name BoardGenerator
|
||
extends RefCounted
|
||
|
||
const METRIC_VERSION := 1
|
||
|
||
# 违约中止通道——缺省空 Callable 退化 push_error+assert（现行为逐字保持）；组装层注入
|
||
# orchestrator._fatal，使 release 剥 assert 后仍经 battle_result 封印真中止。
|
||
var fatal_handler: Callable = Callable()
|
||
|
||
|
||
# ==================== 生成入口 ====================
|
||
|
||
# 目标雷数 N = round(d×(W×H−edge²))（params 快照键：density/w/h/spawn_edge）——布雷定数与产物断言两处的单点公式
|
||
static func target_mine_count(params: Dictionary) -> int:
|
||
return int(round(float(params.density) * float(int(params.w) * int(params.h) - int(params.spawn_edge) * int(params.spawn_edge))))
|
||
|
||
|
||
# rng_stream = 一次 open_stream("board_gen") 实例（一张图的全部取数；重试共用——attempt 边界非 scope 事件）。
|
||
# db 参数不标具体类型原因：冒烟工具注入 FakeDB 桩（calibrate_boards.gd 等）——typed 形参会拒收桩表
|
||
func generate(L: int, D: int, rng_stream: RandomNumberGenerator, db) -> BoardData: # type-hint: ignore（测量面 _BoardRowStub 冒烟夹具鸭子型注入——typed 参拒 stub）
|
||
var row: Dictionary[Variant, Variant] = db.combat.board_gen.resolve_row(L, D)
|
||
if row.is_empty():
|
||
_fatal("BoardGen 行缺失 (%d,%d)" % [L, D])
|
||
return null
|
||
var params := _board_params(row, L, D, int(db.combat.board_gen.spawn_edge)) # 表头直索引——缺成员即崩=fail-fast
|
||
var outcome := _attempt_loop(params, rng_stream, db)
|
||
_finalize(outcome["board"], params, outcome["attempts"], outcome["mode"], outcome["r0_rej"])
|
||
return outcome["board"]
|
||
|
||
|
||
## 生成期表值单次读取快照（防热重载撕裂 attempt 语义；零 RNG）
|
||
func _board_params(row: Dictionary, L: int, D: int, spawn_edge: int) -> Dictionary:
|
||
return {
|
||
"L": L, "D": D,
|
||
"w": int(row.board_width), "h": int(row.board_height),
|
||
"density": float(row.mine_density),
|
||
"target": float(row.inferability_target),
|
||
"solver_rules": String(row.solver_rules),
|
||
"min_safe": int(row.min_safe_count),
|
||
"max_attempts": int(row.max_gen_attempts),
|
||
"board_type": String(row.board_type),
|
||
# 初始展开收窄批（迭代14① T1——两新参数；.get 缺省=legacy 调用方防御口径：shave_r=1 现行投值、
|
||
# r0_cap=0 闸关。正式表行全行投值，加载期校验在 BoardGenTable.validate）
|
||
"shave_r": int(row.get("shave_ring_radius", 1)),
|
||
"r0_cap": int(row.get("r0_max_cap", 0)),
|
||
"spawn_edge": spawn_edge, # 出生区边长（表头单点——经 generate 直索引注入）
|
||
}
|
||
|
||
|
||
## 重试主循环：逐尝试评分择优，产出三分出口结果（board/mode/attempts/r0_rej）
|
||
func _attempt_loop(params: Dictionary, rng_stream: RandomNumberGenerator, db) -> Dictionary: # type-hint: ignore（同 generate——测量面 stub 注入）
|
||
var best := {"legal": null, "legal_score": -1.0, "bad": null, "badness": [999999, 999999, 999999]}
|
||
var attempts_used := 0
|
||
var accepted: BoardData = null
|
||
var accept_mode := ""
|
||
var r0_cap_rejected := 0
|
||
for k in range(1, params.max_attempts + 1):
|
||
attempts_used = k
|
||
var b := _attempt(params, rng_stream, db)
|
||
var badness := _badness(b, params)
|
||
if best["bad"] == null or _lex_less(badness, best["badness"]):
|
||
best["bad"] = b
|
||
best["badness"] = badness
|
||
# R₀≤cap 重掷闸触发计数（迭代14① T3——尝试事实，与合法性判定独立记录；占比=T5 闸触发率读数）
|
||
if int(params.r0_cap) > 0 and b.compute_r0().size() > int(params.r0_cap):
|
||
r0_cap_rejected += 1
|
||
if legality(b, params):
|
||
var inf: float = measure_inferability(b)["inferability"] # 求解器每合法板恰一次（O3 收口——此前重复调用两次）
|
||
if best["legal"] == null or inf > best["legal_score"]:
|
||
best["legal"] = b
|
||
best["legal_score"] = inf
|
||
if inf >= params.target: # 单一判据（棋盘生成：现行规则已回归"可推理率 ≥ 及格线"）
|
||
accepted = b
|
||
accept_mode = "first_pass"
|
||
break
|
||
if accepted == null:
|
||
var fallback := _accept_fallback(params, best["legal"], best["legal_score"], best["bad"])
|
||
accepted = fallback[0]
|
||
accept_mode = fallback[1]
|
||
return {"board": accepted, "mode": accept_mode, "attempts": attempts_used, "r0_rej": r0_cap_rejected}
|
||
|
||
|
||
## 三分出口（重试耗尽）：接受最优（降级标记）或接受最不坏（非法产物标记——玩家永不卡死）
|
||
func _accept_fallback(params: Dictionary, best_legal: BoardData, best_legal_score: float, best_bad: BoardData) -> Array:
|
||
if best_legal != null:
|
||
best_legal.degraded_product = true # 降级标记（独立于 illegal_product——三分出口表要求标记随产物上抛）
|
||
push_warning("[棋盘生成] 重试耗尽无达标——接受最优出图（强告警埋点，target=%.2f best=%.3f）" % [params.target, best_legal_score])
|
||
return [best_legal, "accept_best"]
|
||
best_bad.illegal_product = true
|
||
push_error("[棋盘生成] attempts 耗尽零合法——接受最不坏出图 + 非法产物标记（玩家永不卡死）")
|
||
return [best_bad, "accept_worst"]
|
||
|
||
|
||
# ==================== 五阶段：单次尝试 ====================
|
||
|
||
func _attempt(params: Dictionary, rng_stream: RandomNumberGenerator, db) -> BoardData: # type-hint: ignore（同 generate——测量面 stub 注入）
|
||
var b := BoardData.new()
|
||
b.width = params.w
|
||
b.height = params.h
|
||
b.spawn_edge = int(params.spawn_edge)
|
||
b.spawn_center = Vector2i(int((params.w - 1) / 2.0), int((params.h - 1) / 2.0))
|
||
b.board_type = str(params.board_type)
|
||
b.level_index = params.L
|
||
b.node_depth = params.D
|
||
# ② 布雷：N = round(d×(W×H−edge²)) 先定数；域 = 全格 − 出生区（edge²）；均匀布雷后"先布后删"（_populate_mines 承载——Fisher-Yates 取号段整体随迁）
|
||
var removed := _populate_mines(b, params, rng_stream)
|
||
# ③ 变异布置（地形变异 框架面）：布置约束与算法随首批变异类型定稿（生成期布置接口结构位），
|
||
# 定稿前零布置——清单为空是合法态（P0），静默产出无变异图。
|
||
b.metrics = {
|
||
"metric_version": METRIC_VERSION,
|
||
"ring_removed": removed,
|
||
}
|
||
return b
|
||
|
||
|
||
## 布雷与剃雷域：域 = 全格 − 出生 3×3；均匀布雷（部分 Fisher-Yates，确定性，消费 board_gen 流）
|
||
## 后剃出生环雷，返回 ring_removed（指标语义不变——现值变小=收窄批预期）
|
||
func _populate_mines(b: BoardData, params: Dictionary, rng_stream: RandomNumberGenerator) -> int:
|
||
var spawn := {}
|
||
for c in b.spawn_cells():
|
||
spawn[c] = true
|
||
var domain: Array[Vector2i] = []
|
||
for c in b.all_cells():
|
||
if not spawn.has(c):
|
||
domain.append(c)
|
||
var n_mines: int = target_mine_count(params)
|
||
# 部分 Fisher-Yates（确定性，消费 board_gen 流）——取号序冻结（契约：同种子逐位一致）
|
||
for i in range(0, mini(n_mines, domain.size())):
|
||
var j := rng_stream.randi_range(i, domain.size() - 1)
|
||
var tmp = domain[i]
|
||
domain[i] = domain[j]
|
||
domain[j] = tmp
|
||
for i in range(0, mini(n_mines, domain.size())):
|
||
b.mines[domain[i]] = true
|
||
_compute_digits(b)
|
||
# 先布后删：剃雷域 = 距 spawn_center 切比雪夫距离 ≤1+r 的格（迭代14①收窄批 T2——原「R₀ flood 域∪
|
||
# 外扩 1 环」全域剃雷=开局开阔地巨大根因〔10×10 板 R₀ 均值 68.6=板面 68.6%〕；收窄后开阔地=
|
||
# 出生邻域外的自然 flood。r 读表 shave_ring_radius（r=1 → 5×5=25 格；r=0 即只剃出生 3×3）。
|
||
# 只减不增、无迭代语义不变；零 RNG 消费变化。
|
||
var removed := _shave_ring(b, int(params.get("shave_r", 1)))
|
||
_compute_digits(b)
|
||
return removed
|
||
|
||
|
||
## 剃雷环域（零 RNG）：距出生心切比雪夫 ≤1+r 的环内剃雷，返回剃除数
|
||
func _shave_ring(b: BoardData, shave_r: int) -> int:
|
||
var ring := {}
|
||
for c in b.all_cells():
|
||
if maxi(absi(c.x - b.spawn_center.x), absi(c.y - b.spawn_center.y)) <= (int(b.spawn_edge) - 1) / 2 + shave_r:
|
||
ring[c] = true
|
||
var removed := 0
|
||
for c in ring.keys():
|
||
if b.mines.has(c):
|
||
b.mines.erase(c)
|
||
removed += 1
|
||
return removed
|
||
|
||
|
||
func _compute_digits(b: BoardData) -> void:
|
||
b.digits.clear()
|
||
for c in b.all_cells():
|
||
if b.is_safe(c):
|
||
var d := 0
|
||
for n in b.neighbors8(c):
|
||
if b.is_mine(n):
|
||
d += 1
|
||
b.digits[c] = d
|
||
|
||
|
||
# ==================== F1 可推理率度量（独立入口——与内嵌调用同一实现） ====================
|
||
|
||
# 返回 {r0, r_inf_count, inferability}
|
||
func measure_inferability(b: BoardData) -> Dictionary:
|
||
var safe_set := {}
|
||
for c in b.safe_cells():
|
||
safe_set[c] = true
|
||
var r := {}
|
||
for c in b.compute_r0():
|
||
r[c] = true
|
||
var known_mines := {}
|
||
var changed := true
|
||
while changed:
|
||
changed = false
|
||
# 算子① 连锁：0 格的 8 邻安全格全入 R
|
||
changed = _apply_chain_rule(b, r, safe_set) or changed
|
||
# 信息源：数字约束 single + subset（solver_rules=single_subset 冻结域）
|
||
var constraints := _build_constraints(b, r, known_mines, safe_set)
|
||
changed = _apply_single_rules(constraints, r, known_mines) or changed
|
||
var derived := _derived_constraints(constraints)
|
||
changed = _apply_derived_rules(derived, r, known_mines) or changed
|
||
var r_inf := r.keys()
|
||
var s_count := safe_set.size()
|
||
var inferability: float = float(r_inf.size()) / float(maxi(1, s_count))
|
||
return {
|
||
"r0": b.compute_r0().size(),
|
||
"r_inf_count": r_inf.size(),
|
||
"inferability": inferability,
|
||
}
|
||
|
||
|
||
## 算子① 连锁域：0 格的 8 邻安全格全入 R（返回本轮是否有新增）
|
||
func _apply_chain_rule(b: BoardData, r: Dictionary, safe_set: Dictionary) -> bool:
|
||
var changed := false
|
||
var frontier: Array[Variant] = r.keys()
|
||
for x in frontier:
|
||
if safe_set.has(x) and b.digit_of(x) == 0:
|
||
for n in b.neighbors8(x):
|
||
if safe_set.has(n) and not r.has(n):
|
||
r[n] = true
|
||
changed = true
|
||
return changed
|
||
|
||
|
||
## single 规则域：零雷约束开格／满雷约束标雷（返回本轮是否有新增）
|
||
func _apply_single_rules(constraints: Array, r: Dictionary, known_mines: Dictionary) -> bool:
|
||
var changed := false
|
||
for con in constraints:
|
||
if con.mines == 0:
|
||
for c in con.unknown:
|
||
if not r.has(c):
|
||
r[c] = true
|
||
changed = true
|
||
elif con.mines == con.unknown.size():
|
||
for c in con.unknown:
|
||
if not known_mines.has(c):
|
||
known_mines[c] = true
|
||
changed = true
|
||
return changed
|
||
|
||
|
||
## subset 派生域：U1 ⊂ U2 ⇒ (d2−d1, U2∖U1)（对约束对做包含判定，产出派生约束集）
|
||
func _derived_constraints(constraints: Array) -> Array:
|
||
var derived := []
|
||
for i in constraints.size():
|
||
for j in range(i + 1, constraints.size()):
|
||
var a: Dictionary[Variant, Variant] = constraints[i]
|
||
var bb: Dictionary[Variant, Variant] = constraints[j]
|
||
var sub := _subset_of(a.unknown, bb.unknown)
|
||
if sub == 1:
|
||
derived.append(_diff_constraint(a, bb))
|
||
elif sub == 2:
|
||
derived.append(_diff_constraint(bb, a))
|
||
return derived
|
||
|
||
|
||
## 派生规则域：派生约束的零雷开格／满雷标雷（同 no-peek 口径：推导结论不再用 safe_set 复核——约束本身可靠）
|
||
func _apply_derived_rules(derived: Array, r: Dictionary, known_mines: Dictionary) -> bool:
|
||
var changed := false
|
||
for con in derived:
|
||
if con.mines == 0:
|
||
for c in con.unknown:
|
||
if not r.has(c):
|
||
r[c] = true
|
||
changed = true
|
||
elif con.mines == con.unknown.size():
|
||
for c in con.unknown:
|
||
if not known_mines.has(c):
|
||
known_mines[c] = true
|
||
changed = true
|
||
return changed
|
||
|
||
|
||
func _build_constraints(b: BoardData, r: Dictionary, known_mines: Dictionary, safe_set: Dictionary) -> Array:
|
||
var out: Array[Dictionary] = []
|
||
for c in r.keys():
|
||
if not safe_set.has(c):
|
||
continue # 只对已开安全格的数字建约束
|
||
var d := b.digit_of(c)
|
||
var unknown := {}
|
||
var known_here := 0
|
||
for n in b.neighbors8(c):
|
||
if known_mines.has(n):
|
||
known_here += 1
|
||
elif not r.has(n):
|
||
unknown[n] = true # 未揭示格一律计入未知（含危险格）——求解器禁读 safe_set 真值（棋盘生成：雷位不许偷看）
|
||
var d_eff := d - known_here
|
||
if d_eff < 0 or (d_eff > 0 and unknown.is_empty()):
|
||
continue # 矛盾态（非法板）——不产约束，验收由 legality 闸承担
|
||
if not unknown.is_empty():
|
||
out.append({"mines": d_eff, "unknown": unknown})
|
||
return out
|
||
|
||
|
||
func _subset_of(a: Dictionary, b: Dictionary) -> int:
|
||
# 1: a ⊂ b；2: b ⊂ a；0: 互不包含
|
||
var a_in_b := true
|
||
var b_in_a := true
|
||
for k in a.keys():
|
||
if not b.has(k):
|
||
a_in_b = false
|
||
for k in b.keys():
|
||
if not a.has(k):
|
||
b_in_a = false
|
||
if a.size() < b.size() and a_in_b:
|
||
return 1
|
||
if b.size() < a.size() and b_in_a:
|
||
return 2
|
||
return 0
|
||
|
||
|
||
func _diff_constraint(big: Dictionary, small: Dictionary) -> Dictionary:
|
||
var diff := {}
|
||
for k in big.unknown.keys():
|
||
if not small.unknown.has(k):
|
||
diff[k] = true
|
||
return {"mines": big.mines - small.mines, "unknown": diff}
|
||
|
||
|
||
# ==================== 合法性闸（独立入口；F5 legality + R₀≠S 修订义务） ====================
|
||
|
||
func is_clearable(b: BoardData) -> bool:
|
||
var safe_set := {}
|
||
for c in b.safe_cells():
|
||
safe_set[c] = true
|
||
var seen := {}
|
||
var queue: Array[Variant] = b.spawn_cells().duplicate()
|
||
for c in queue:
|
||
seen[c] = true
|
||
while not queue.is_empty():
|
||
var c: Vector2i = queue.pop_front()
|
||
if not safe_set.has(c):
|
||
continue
|
||
for n in b.neighbors8(c):
|
||
if safe_set.has(n) and not seen.has(n):
|
||
seen[n] = true
|
||
queue.append(n)
|
||
for c in safe_set.keys():
|
||
if not seen.has(c):
|
||
return false
|
||
return true
|
||
|
||
|
||
func legality(b: BoardData, params: Dictionary) -> bool:
|
||
var reasons := _legality_faults(b, params)
|
||
return reasons.is_empty()
|
||
|
||
|
||
func _legality_faults(b: BoardData, params: Dictionary) -> Array:
|
||
var faults: Array[String] = []
|
||
# ① 出生区无雷 ② 出生区∪R₀ 无变异 ③ 数字恒真 ④ R₀ ≠ S（排雷玩法 修订义务）
|
||
for c in b.spawn_cells():
|
||
if b.is_mine(c):
|
||
faults.append("inv1_mine_in_spawn")
|
||
var r0 := {}
|
||
for c in b.compute_r0():
|
||
r0[c] = true
|
||
if b.variant_ids(c).size() > 0:
|
||
faults.append("inv2b_variant_in_r0")
|
||
for c in b.spawn_cells():
|
||
if b.variant_ids(c).size() > 0:
|
||
faults.append("inv2a_variant_in_spawn")
|
||
for c in b.all_cells():
|
||
if b.is_safe(c) and b.digit_of(c) != _neighbor_mine_count(b, c):
|
||
faults.append("inv3_digit_invariant")
|
||
if r0.size() == b.safe_cells().size():
|
||
faults.append("inv4_r0_equals_s")
|
||
# clearable（合法性闸——与质量度量两闸分离）
|
||
if not is_clearable(b):
|
||
faults.append("not_clearable")
|
||
# min_safe 产物侧（容错两种入参形态：generate 的 params 快照 / 原始表行）
|
||
var min_safe: int = int(params.get("min_safe", params.get("min_safe_count", 0)))
|
||
if b.safe_cells().size() < min_safe:
|
||
faults.append("min_safe_short")
|
||
# ⑤ R₀≤cap 重掷闸（迭代14①收窄批 T3——合法性先于质量评分，同 inv4 白给板先例；超阈整板作废
|
||
# 同流重试〔重试共用 board_gen 流，attempt 边界非 scope 事件〕；全部尝试超阈 → 既有三分出口
|
||
# accept_worst 诚实降级+非法产物标记（玩家永不卡死）。cap=0 = 闸关（legacy 调用方防御口径）
|
||
var r0_cap: int = int(params.get("r0_cap", 0))
|
||
if r0_cap > 0 and r0.size() > r0_cap:
|
||
faults.append("r0_cap_exceeded")
|
||
return faults
|
||
|
||
|
||
func _neighbor_mine_count(b: BoardData, c: Vector2i) -> int:
|
||
var d := 0
|
||
for n in b.neighbors8(c):
|
||
if b.is_mine(n):
|
||
d += 1
|
||
return d
|
||
|
||
|
||
# ==================== 产出上报（生成期指标集） ====================
|
||
|
||
func _finalize(b: BoardData, params: Dictionary, attempts_used: int, accept_mode: String, r0_cap_rejected: int = 0) -> void:
|
||
var m := measure_inferability(b)
|
||
# 产物断言（棋盘生成 出生区安全与反退化布雷）：实际雷数 ≤ 目标雷数 N——先布后删只减不增，失败即致命
|
||
var n_target: int = target_mine_count(params)
|
||
if b.mines.size() > n_target:
|
||
_fatal("产物断言违例：实际雷数 %d > 目标雷数 N=%d（先布后删只减不增）" % [b.mines.size(), n_target])
|
||
var attempt_facts := b.metrics # 保留尝试期事实（ring_removed）
|
||
b.metrics = {
|
||
"metric_version": METRIC_VERSION,
|
||
"inferability": m.inferability,
|
||
"actual_mines": b.mines.size(),
|
||
"target_mines": n_target,
|
||
"actual_safe_count": b.safe_cells().size(),
|
||
"attempts_used": attempts_used,
|
||
"accept_mode": accept_mode,
|
||
"degraded_product": accept_mode == "accept_best",
|
||
"illegal_product": accept_mode == "accept_worst",
|
||
"ring_removed": attempt_facts.get("ring_removed", 0),
|
||
"r0_count": m["r0"],
|
||
# R₀≤cap 重掷闸尝试事实（迭代14① T3——占比=T5 闸触发率读数，过高=甲乙配比失衡信号）
|
||
"r0_cap_rejected": r0_cap_rejected,
|
||
}
|
||
b.illegal_product = accept_mode == "accept_worst"
|
||
b.degraded_product = accept_mode == "accept_best"
|
||
|
||
|
||
func _badness(b: BoardData, params: Dictionary) -> Array:
|
||
# F5 k*** 确定性劣度字典序：(不可达安全格数, min_safe 缺口, 序号)
|
||
var unreachable := 0
|
||
var safe_set := {}
|
||
for c in b.safe_cells():
|
||
safe_set[c] = true
|
||
var seen := {}
|
||
var queue: Array[Variant] = b.spawn_cells().duplicate()
|
||
for c in queue:
|
||
seen[c] = true
|
||
while not queue.is_empty():
|
||
var c: Vector2i = queue.pop_front()
|
||
if not safe_set.has(c):
|
||
continue
|
||
for n in b.neighbors8(c):
|
||
if safe_set.has(n) and not seen.has(n):
|
||
seen[n] = true
|
||
queue.append(n)
|
||
for c in safe_set.keys():
|
||
if not seen.has(c):
|
||
unreachable += 1
|
||
var gap: int = maxi(0, int(params.min_safe) - b.safe_cells().size())
|
||
return [unreachable, gap, 0]
|
||
|
||
|
||
func _lex_less(a: Array, b: Array) -> bool:
|
||
for i in a.size():
|
||
if a[i] < b[i]:
|
||
return true
|
||
if a[i] > b[i]:
|
||
return false
|
||
return false
|
||
|
||
|
||
func _fatal(msg: String) -> void:
|
||
push_error("[棋盘生成/fatal] " + msg)
|
||
if fatal_handler.is_valid():
|
||
fatal_handler.call(msg) # 组装层注入 orchestrator._fatal（封印传导，不绕第二中止通道）
|
||
assert(false, msg)
|