commit
96353480be
|
|
@ -1,2 +1,37 @@
|
|||
./idea/
|
||||
.idea/workspace.xml
|
||||
|
||||
|
||||
# Python 缓存文件
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
|
||||
# 虚拟环境
|
||||
.venv/
|
||||
|
||||
# IDE 文件
|
||||
.idea/
|
||||
.vscode/
|
||||
*.swp
|
||||
*.swo
|
||||
|
||||
# 日志和调试文件
|
||||
*.log
|
||||
debug.log
|
||||
|
||||
|
||||
# 单元测试覆盖率报告
|
||||
htmlcov/
|
||||
.coverage
|
||||
*.cover
|
||||
.coverage.*
|
||||
.cache
|
||||
nosetests.xml
|
||||
coverage.xml
|
||||
*.mocha.json
|
||||
test-results/
|
||||
|
||||
|
||||
# 机器学习项目中的临时文件
|
||||
logs/
|
||||
109
README.md
109
README.md
|
|
@ -28,8 +28,6 @@
|
|||
|
||||
#### **5. 核心代码目录 (`src/`)**
|
||||
|
||||
#### **5. 核心代码目录 (`src/`)**
|
||||
|
||||
##### **游戏引擎 (`engine/`)**
|
||||
|
||||
- `mahjong_engine.py`：实现成都麻将规则，包括摸牌、打牌、碰、杠、胡牌等逻辑。
|
||||
|
|
@ -94,26 +92,117 @@
|
|||
#### 详细番数计算
|
||||
|
||||
1. **平胡（基本胡）**：四坎牌加一对将，四坎牌可以是刻子或顺子，计为1番。
|
||||
|
||||
2. **清一色**：
|
||||
|
||||
- 不带杠的清一色称为“素清”，计为2番。
|
||||
- 带杠的清一色或清一色对子胡（简称“清对”）计为3番，称为“极品”，点炮40分。
|
||||
- 带两杠的清一色或清一色对子胡带杠计为4番，称为“极中极”或“精品”，点炮80分。
|
||||
3. **带幺九**：手牌中含有1或9的牌，计为3番。
|
||||
- 带杠的清一色或清一色对子胡（简称“清对”）计为3番，称为“极品”。
|
||||
- 带两杠的清一色或清一色对子胡带杠计为4番，称为“极中极”或“精品”。<!--存疑-->
|
||||
|
||||
3. **带幺九**：
|
||||
|
||||
- **带幺九**：指玩家手上的牌全部是由1和9组成的顺子、刻子或对子。例如，123, 789, 111, 999, 11等。计为3番。
|
||||
|
||||
- **清带幺九**：指玩家手上的牌不仅全部由1和9组成，而且是同一花色（条、筒、万），即清一色的带幺九。计为1番。<!--存疑-->
|
||||
|
||||
4. **七对**：手牌由7个对子组成，计为2番。
|
||||
|
||||
5. **全求人**：所有牌都是通过碰、杠、吃别人打出的牌来完成的，计为6番。
|
||||
|
||||
6. **龙七对**：七对中有一对是三张相同的牌，计为12番。
|
||||
|
||||
7. **清七对**：全部由一种花色组成的七对，计为12番。
|
||||
|
||||
8. **杠上开花**：在杠牌之后立即自摸胡牌，计为1番。
|
||||
|
||||
9. **抢杠胡**：当其他玩家明杠时，你正好可以胡那张牌，计为1番。
|
||||
|
||||
10. **天胡**：庄家起牌后直接胡牌，计为12番。
|
||||
|
||||
11. **地胡**：闲家在第一轮打牌时就胡牌，计为12番。
|
||||
|
||||
12. **大对子**：手牌由四个对子加一个刻子组成，计为2番。
|
||||
|
||||
13. **小七对**：有六对加上一个对子，计为2番。
|
||||
14. **杠上炮**：在杠牌之后放炮让他人胡牌，通常不加分，但有时会根据地方规则有所调整。
|
||||
15. **金钩吊**：手上只剩下一张牌等别人打出，然后胡牌，计为1番。
|
||||
16. **海底捞月**：最后一张牌被玩家摸到并胡牌，计为1番。
|
||||
17. **海底炮**：最后一张牌被打出，导致玩家胡牌，计为1番。
|
||||
|
||||
14. **金钩吊**：手上只剩下一张牌等别人打出，然后胡牌，计为1番。
|
||||
|
||||
15. **海底捞月**：最后一张牌被玩家摸到并胡牌，计为1番。
|
||||
|
||||
16. **海底炮**：最后一张牌被打出，导致玩家胡牌，计为1番。
|
||||
|
||||
这些番数可以叠加，例如，如果一个玩家同时满足了清一色和七对，那么他的总番数就是2番（清一色）+ 2番（七对）= 4番。
|
||||
|
||||
## 成都麻将规则建模
|
||||
### 计分规则
|
||||
|
||||
- **基本分值**：每番的具体分值可以根据不同的玩法和地区有所不同，但一般情况下，每番的分值可以设定为一个固定的数值，比如5分、10分等。
|
||||
- **翻倍规则**：某些地方可能会有额外的翻倍规则，例如，如果胡牌者是在“海底捞月”或“杠上开花”等特殊情况下胡牌，可能会有额外的加分。
|
||||
|
||||
**自摸 **：是指玩家通过自己摸牌完成胡牌。自摸时，其他玩家都需要给赢家支付相应的分数。
|
||||
|
||||
## 成都麻将规则建模
|
||||
|
||||
麻将游戏引擎建模代码于项目根src/engine/目录下。
|
||||
|
||||
## PPO（Proximal Policy Optimization）算法
|
||||
|
||||
TensorBoard 通常会记录和可视化多种训练指标。你提到的这些图表代表了 PPO 训练过程中的不同方面。下面是对每个图表的解释：
|
||||
|
||||
### 1. `train/loss`
|
||||
|
||||
- **含义**：总损失函数值。
|
||||
- **用途**：这是所有损失项的综合，包括策略梯度损失、价值函数损失等。通过观察这个指标，可以了解整个训练过程中的总体损失趋势。
|
||||
|
||||
### 2. `train/policy_gradient_loss`
|
||||
|
||||
- **含义**：策略梯度损失。
|
||||
- **用途**：这表示策略网络更新时的损失。PPO 通过剪裁来限制策略更新的幅度，以确保稳定的学习过程。观察这个指标可以帮助你了解策略更新的情况。
|
||||
|
||||
### 3. `train/value_loss`
|
||||
|
||||
- **含义**：价值函数损失。
|
||||
- **用途**：这表示价值网络（用于估计状态值或状态-动作对的价值）的损失。价值函数的准确性对于评估策略的好坏非常重要。通过观察这个指标，可以了解价值网络的学习情况。
|
||||
|
||||
### 4. `train/learning_rate`
|
||||
|
||||
- **含义**：当前的学习率。
|
||||
- **用途**：学习率是优化器的一个重要超参数，控制着每次更新时权重调整的幅度。观察学习率的变化可以帮助你了解学习率调度策略的效果。
|
||||
|
||||
### 5. `train/explained_variance`
|
||||
|
||||
- **含义**：解释方差。
|
||||
- **用途**：这是一个衡量价值函数预测与实际回报之间差异的指标。解释方差越接近 1，说明价值函数的预测越准确。通过观察这个指标，可以评估价值函数的性能。
|
||||
|
||||
### 6. `train/entropy_loss`
|
||||
|
||||
- **含义**：熵损失。
|
||||
- **用途**：熵损失鼓励策略具有一定的随机性，以防止过早收敛到局部最优解。通过观察这个指标，可以了解策略的探索程度。
|
||||
|
||||
### 7. `train/clip_range`
|
||||
|
||||
- **含义**：剪裁范围。
|
||||
- **用途**：PPO 使用剪裁来限制策略更新的幅度，以确保稳定性。剪裁范围是一个重要的超参数，决定了剪裁的严格程度。观察这个指标可以帮助你了解剪裁策略的效果。
|
||||
|
||||
### 8. `train/clip_fraction`
|
||||
|
||||
- **含义**：被剪裁的比例。
|
||||
- **用途**：这个指标表示有多少比例的更新被剪裁。如果剪裁比例很高，可能意味着你的策略更新过于激进，需要调整剪裁范围或其他超参数。
|
||||
|
||||
### 9. `train/approx_kl`
|
||||
|
||||
- **含义**：近似 KL 散度。
|
||||
- **用途**：KL 散度衡量新旧策略之间的差异。PPO 通过控制 KL 散度来确保策略更新的稳定性。通过观察这个指标，可以了解策略更新的幅度和稳定性。
|
||||
|
||||
### 总结
|
||||
|
||||
这些图表提供了关于 PPO 训练过程的全面视图，帮助你监控和调试模型。以下是每个图表的主要用途：
|
||||
|
||||
- **`train/loss`**：总体损失，反映训练的整体进展。
|
||||
- **`train/policy_gradient_loss`**：策略梯度损失，反映策略网络的更新情况。
|
||||
- **`train/value_loss`**：价值函数损失，反映价值网络的学习情况。
|
||||
- **`train/learning_rate`**：学习率，反映优化器的设置。
|
||||
- **`train/explained_variance`**：解释方差，反映价值函数的准确性。
|
||||
- **`train/entropy_loss`**：熵损失，反映策略的探索程度。
|
||||
- **`train/clip_range`**：剪裁范围，反映策略更新的限制。
|
||||
- **`train/clip_fraction`**：被剪裁的比例，反映策略更新的稳定性。
|
||||
- **`train/approx_kl`**：近似 KL 散度，反映策略更新的幅度和稳定性。
|
||||
|
|
@ -0,0 +1,14 @@
|
|||
# configs/log_config.py
|
||||
from loguru import logger
|
||||
import os
|
||||
|
||||
def setup_logging():
|
||||
# 确保日志目录存在
|
||||
log_dir = "../logs"
|
||||
os.makedirs(log_dir, exist_ok=True)
|
||||
|
||||
# 清除所有现有日志处理器，防止重复配置
|
||||
logger.remove()
|
||||
|
||||
# 配置日志，记录到 ../logs 目录下
|
||||
logger.add(os.path.join(log_dir, "chengdu_mj_engine.log"), rotation="10 MB", level="DEBUG", format="{time} {level} {message}")
|
||||
Binary file not shown.
|
|
@ -0,0 +1,4 @@
|
|||
loguru~=0.7.2
|
||||
pytest~=8.3.3
|
||||
gym~=0.26.2
|
||||
numpy~=2.1.3
|
||||
|
|
@ -0,0 +1,35 @@
|
|||
import gym
|
||||
from stable_baselines3 import PPO
|
||||
from src.environment.chengdu_majiang_env import MahjongEnv
|
||||
import torch
|
||||
from configs.log_config import setup_logging
|
||||
|
||||
def train_model():
|
||||
# 创建 MahjongEnv 环境实例
|
||||
env = MahjongEnv()
|
||||
|
||||
# 检查是否有可用的GPU
|
||||
device = "cuda" if torch.cuda.is_available() else "cpu"
|
||||
print(f"使用设备: {device}")
|
||||
|
||||
# 使用 PPO 算法训练模型
|
||||
model = PPO("MlpPolicy", env, verbose=1, tensorboard_log="../logs/ppo_mahjong_tensorboard/", device=device)
|
||||
|
||||
# 训练模型，训练总步数为100000
|
||||
model.learn(total_timesteps=100)
|
||||
|
||||
# 保存训练后的模型
|
||||
model.save("../models/ppo_mahjong_model")
|
||||
|
||||
# 测试模型
|
||||
obs = env.reset()
|
||||
done = False
|
||||
while not done:
|
||||
action, _states = model.predict(obs) # 使用训练好的模型来选择动作
|
||||
obs, reward, done, info = env.step(action) # 执行动作
|
||||
env.render() # 打印环境状态
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 调用配置函数来设置日志
|
||||
setup_logging()
|
||||
train_model()
|
||||
|
|
@ -1,13 +1,124 @@
|
|||
def draw_tile(state):
|
||||
if state.remaining_tiles == 0:
|
||||
raise ValueError("牌堆已空")
|
||||
tile = state.deck.pop(0)
|
||||
state.remaining_tiles -= 1
|
||||
state.hands[state.current_player][tile] += 1
|
||||
return tile
|
||||
from loguru import logger
|
||||
from src.engine.utils import get_tile_name
|
||||
|
||||
def discard_tile(state, tile):
|
||||
if state.hands[state.current_player][tile] == 0:
|
||||
|
||||
def draw_tile(engine):
|
||||
"""
|
||||
当前玩家摸牌逻辑，记录牌的详细信息和游戏状态。
|
||||
"""
|
||||
if engine.state.remaining_tiles == 0:
|
||||
logger.warning("牌堆已空，游戏结束！")
|
||||
engine.game_over = True
|
||||
return 0, True # 游戏结束时返回 0 和 done = True
|
||||
|
||||
tile = engine.state.deck.pop(0) # 从牌堆中取出一张牌
|
||||
engine.state.remaining_tiles -= 1 # 更新剩余牌数
|
||||
engine.state.hands[engine.state.current_player][tile] += 1 # 加入当前玩家手牌
|
||||
|
||||
tile_name = get_tile_name(tile) # 获取具体的牌名
|
||||
logger.info(
|
||||
f"玩家 {engine.state.current_player} 摸到一张牌: {tile_name}（索引 {tile}）。剩余牌堆数量: {engine.state.remaining_tiles}"
|
||||
)
|
||||
|
||||
# 返回奖励和游戏是否结束的标志
|
||||
return 0, False # 奖励为 0，done 为 False（游戏继续）
|
||||
|
||||
|
||||
def discard_tile(self, tile):
|
||||
"""
|
||||
当前玩家打牌逻辑，记录打出的牌和当前牌河信息。
|
||||
"""
|
||||
if self.state.hands[self.state.current_player][tile] == 0:
|
||||
logger.error(f"玩家 {self.state.current_player} 尝试打出不存在的牌: 索引 {tile}")
|
||||
raise ValueError("玩家没有这张牌")
|
||||
state.hands[state.current_player][tile] -= 1
|
||||
state.discards[state.current_player].append(tile)
|
||||
|
||||
self.state.hands[self.state.current_player][tile] -= 1 # 从手牌中移除
|
||||
self.state.discards[self.state.current_player].append(tile) # 加入牌河
|
||||
|
||||
tile_name = get_tile_name(tile) # 获取具体的牌名
|
||||
logger.info(
|
||||
f"玩家 {self.state.current_player} 打出一张牌: {tile_name}（索引 {tile}）。当前牌河: {[get_tile_name(t) for t in self.state.discards[self.state.current_player]]}"
|
||||
)
|
||||
|
||||
|
||||
def peng(self, tile):
|
||||
"""
|
||||
当前玩家碰牌逻辑，记录碰牌操作和手牌状态。
|
||||
"""
|
||||
player = self.state.current_player
|
||||
if self.state.hands[player][tile] < 2:
|
||||
logger.error(f"玩家 {player} 尝试碰牌失败: {get_tile_name(tile)}（索引 {tile}）")
|
||||
raise ValueError("碰牌条件不满足")
|
||||
|
||||
self.state.hands[player][tile] -= 2 # 减去两张牌
|
||||
self.state.melds[player].append(("peng", tile)) # 加入明牌列表
|
||||
|
||||
tile_name = get_tile_name(tile)
|
||||
logger.info(f"玩家 {player} 碰了一张牌: {tile_name}（索引 {tile}）。当前明牌: {self.state.melds[player]}")
|
||||
|
||||
|
||||
def gang(self, tile, mode):
|
||||
"""
|
||||
当前玩家杠牌逻辑，记录杠牌类型和状态更新。
|
||||
"""
|
||||
player = self.state.current_player
|
||||
tile_name = get_tile_name(tile)
|
||||
|
||||
if mode == "ming" and self.state.hands[player][tile] == 3:
|
||||
self.state.hands[player][tile] -= 3
|
||||
self.state.melds[player].append(("ming_gang", tile))
|
||||
logger.info(f"玩家 {player} 明杠: {tile_name}（索引 {tile}）")
|
||||
self.state.scores[player] += 1 # 奖励1分
|
||||
logger.info(f"玩家 {player} 因明杠获得1分")
|
||||
|
||||
elif mode == "an" and self.state.hands[player][tile] == 4:
|
||||
self.state.hands[player][tile] -= 4
|
||||
self.state.melds[player].append(("an_gang", tile))
|
||||
logger.info(f"玩家 {player} 暗杠: {tile_name}（索引 {tile}）")
|
||||
self.state.scores[player] += 1 # 奖励1分
|
||||
logger.info(f"玩家 {player} 因暗杠获得1分")
|
||||
|
||||
else:
|
||||
logger.error(f"玩家 {player} 尝试杠牌失败: {tile_name}（索引 {tile}），条件不满足")
|
||||
raise ValueError("杠牌条件不满足")
|
||||
|
||||
|
||||
|
||||
def check_blood_battle(self):
|
||||
"""
|
||||
检查游戏是否流局或血战结束，记录状态。
|
||||
"""
|
||||
if any(score <= 0 for score in self.state.scores):
|
||||
logger.info(f"游戏结束，某玩家分数小于等于0: {self.state.scores}")
|
||||
self.game_over = True
|
||||
|
||||
if len(self.state.winners) >= 3 or self.state.remaining_tiles == 0:
|
||||
logger.info(f"游戏结束，赢家列表: {self.state.winners}")
|
||||
self.game_over = True
|
||||
|
||||
|
||||
def set_missing_suit(player, missing_suit, game_state):
|
||||
"""
|
||||
玩家设置缺门的动作。
|
||||
|
||||
参数:
|
||||
- player: 玩家索引（0-3）。
|
||||
- missing_suit: 玩家选择的缺门（"条"、"筒" 或 "万"）。
|
||||
- game_state: 当前的游戏状态（`ChengduMahjongState` 实例）。
|
||||
|
||||
异常:
|
||||
- ValueError: 如果缺门设置无效。
|
||||
"""
|
||||
valid_suits = ["条", "筒", "万"]
|
||||
if missing_suit not in valid_suits:
|
||||
logger.error(f"玩家 {player} 尝试设置无效的缺门: {missing_suit}")
|
||||
raise ValueError("缺门设置无效")
|
||||
|
||||
if game_state.missing_suits[player] is not None:
|
||||
logger.error(f"玩家 {player} 已经设置了缺门，不能重复设置")
|
||||
raise ValueError("缺门已经设置，不能重复设置")
|
||||
|
||||
game_state.missing_suits[player] = missing_suit
|
||||
logger.info(f"玩家 {player} 设置缺门为: {missing_suit}")
|
||||
|
||||
return game_state.missing_suits[player]
|
||||
|
|
@ -0,0 +1,98 @@
|
|||
def calculate_fan(hand, melds, is_self_draw, is_cleared, conditions):
|
||||
"""
|
||||
根据规则动态计算番数。
|
||||
|
||||
参数:
|
||||
- hand: 当前胡牌的手牌（长度为108的列表，表示每张牌的数量）。
|
||||
- melds: 碰杠等明牌列表。
|
||||
- is_self_draw: 是否自摸。
|
||||
- is_cleared: 是否清一色。
|
||||
- conditions: 其他胡牌条件的字典，例如 {'is_seven_pairs': True, 'add_self_draw': True}。
|
||||
|
||||
返回:
|
||||
- fan: 总番数。
|
||||
"""
|
||||
fan = 0 # 初始番数
|
||||
|
||||
# 定义番种规则
|
||||
rules = {
|
||||
"is_pure_cleared": lambda: 3 if is_cleared and len(melds) == 3 and not conditions.get("is_double_pure_cleared",
|
||||
False) else 0,
|
||||
"is_double_pure_cleared": lambda: 4 if is_cleared and len(melds) >= 2 and conditions.get(
|
||||
"is_double_pure_cleared", False) else 0,
|
||||
"is_cleared": lambda: 2 if is_cleared and not (conditions.get("is_pure_cleared", False) or
|
||||
conditions.get("is_double_pure_cleared", False) or
|
||||
conditions.get("is_clear_seven_pairs", False)) else 0,
|
||||
"is_seven_pairs": lambda: 2 if conditions.get("is_seven_pairs", False) and not conditions.get(
|
||||
"is_dragon_seven_pairs", False) else 0,
|
||||
"is_dragon_seven_pairs": lambda: 12 if conditions.get("is_dragon_seven_pairs", False) else 0,
|
||||
"is_clear_seven_pairs": lambda: 12 if conditions.get("is_clear_seven_pairs", False) else 0,
|
||||
"is_big_pairs": lambda: 2 if conditions.get("is_big_pairs", False) else 0,
|
||||
"is_small_pairs": lambda: 2 if conditions.get("is_small_pairs", False) else 0,
|
||||
"is_full_request": lambda: 6 if conditions.get("is_full_request", False) else 0,
|
||||
"is_gang_flower": lambda: 1 if conditions.get("is_gang_flower", False) else 0,
|
||||
"is_rob_gang": lambda: 1 if conditions.get("is_rob_gang", False) else 0,
|
||||
"is_under_the_sea": lambda: 1 if conditions.get("is_under_the_sea", False) else 0,
|
||||
"is_tian_hu": lambda: 12 if conditions.get("is_tian_hu", False) else 0,
|
||||
"is_di_hu": lambda: 12 if conditions.get("is_di_hu", False) else 0,
|
||||
"basic_win": lambda: 1 if not (conditions.get("is_seven_pairs", False) or
|
||||
conditions.get("is_big_pairs", False) or
|
||||
conditions.get("is_dragon_seven_pairs", False) or
|
||||
conditions.get("is_pure_cleared", False) or
|
||||
conditions.get("is_double_pure_cleared", False) or
|
||||
conditions.get("is_small_pairs", False) or
|
||||
conditions.get("is_clear_seven_pairs", False) or
|
||||
conditions.get("is_full_request", False) or
|
||||
conditions.get("is_rob_gang", False) or
|
||||
conditions.get("is_under_the_sea", False) or
|
||||
conditions.get("is_tian_hu", False) or
|
||||
conditions.get("is_di_hu", False)) else 0,
|
||||
|
||||
}
|
||||
|
||||
print("\nCalculating fan...")
|
||||
# 逐一应用规则
|
||||
for rule, func in rules.items():
|
||||
result = func()
|
||||
fan += result
|
||||
print(f"Rule: {rule}, Fan: {result}") # 调试输出
|
||||
|
||||
return fan
|
||||
|
||||
|
||||
def is_seven_pairs(hand):
|
||||
"""
|
||||
检查手牌是否是七对。
|
||||
"""
|
||||
return sum(1 for count in hand if count == 2) == 7
|
||||
|
||||
|
||||
def is_cleared(hand, melds):
|
||||
"""
|
||||
检查手牌和明牌是否是清一色。
|
||||
|
||||
参数:
|
||||
- hand: 当前胡牌的手牌（长度为108的列表，表示每张牌的数量）。
|
||||
- melds: 碰杠等明牌列表。
|
||||
|
||||
返回:
|
||||
- bool: 是否为清一色。
|
||||
"""
|
||||
# 获取所有牌的花色
|
||||
all_tiles = hand + [tile for meld in melds for tile in meld]
|
||||
suits = [tile // 36 for tile in all_tiles if tile > 0]
|
||||
|
||||
# 检查是否有多种花色
|
||||
return len(set(suits)) == 1
|
||||
|
||||
def is_big_pairs(hand):
|
||||
"""
|
||||
检查手牌是否是大对子（由刻子和一对组成）。
|
||||
"""
|
||||
from collections import Counter
|
||||
counter = Counter(hand)
|
||||
|
||||
counts = [count for count in hand if count > 0]
|
||||
result = counts.count(2) == 1 and counts.count(3) >= 3
|
||||
print(f"Big pairs check: {result}, Counter: {Counter(counts)}")
|
||||
return result
|
||||
|
|
@ -1,89 +1,47 @@
|
|||
import random
|
||||
|
||||
from loguru import logger
|
||||
|
||||
from .game_state import ChengduMahjongState
|
||||
|
||||
|
||||
class ChengduMahjongEngine:
|
||||
def __init__(self):
|
||||
self.state = ChengduMahjongState()
|
||||
self.state = ChengduMahjongState() # 创建游戏状态
|
||||
self.game_over = False
|
||||
self.game_started = False # 游戏是否已开始
|
||||
self.deal_tiles() # 发牌
|
||||
|
||||
def draw_tile(self):
|
||||
# 当前玩家摸牌
|
||||
if self.state.remaining_tiles == 0:
|
||||
self.game_over = True
|
||||
return "牌堆已空"
|
||||
tile = self.state.deck.pop(0)
|
||||
self.state.remaining_tiles -= 1
|
||||
self.state.hands[self.state.current_player][tile] += 1
|
||||
return tile
|
||||
def deal_tiles(self):
|
||||
""" 发牌，每个玩家发13张牌，并设置缺门 """
|
||||
logger.info("发牌中...")
|
||||
|
||||
def discard_tile(self, tile):
|
||||
# 当前玩家打牌
|
||||
if self.state.hands[self.state.current_player][tile] == 0:
|
||||
raise ValueError("当前玩家没有这张牌")
|
||||
self.state.hands[self.state.current_player][tile] -= 1
|
||||
self.state.discards[self.state.current_player].append(tile)
|
||||
# 洗牌（随机打乱牌堆）
|
||||
random.shuffle(self.state.deck)
|
||||
|
||||
def peng(self, tile):
|
||||
# 碰牌逻辑
|
||||
player = self.state.current_player
|
||||
if self.state.hands[player][tile] < 2:
|
||||
raise ValueError("碰牌条件不满足")
|
||||
self.state.hands[player][tile] -= 2
|
||||
self.state.melds[player].append(("peng", tile))
|
||||
# 随机发牌给每个玩家
|
||||
for player in range(4):
|
||||
for _ in range(13): # 每个玩家13张牌
|
||||
tile = self.state.deck.pop() # 从牌堆抽取一张牌
|
||||
self.state.hands[player][tile] += 1 # 增加玩家手牌的计数
|
||||
|
||||
def gang(self, tile, mode="ming"):
|
||||
# 杠牌逻辑
|
||||
player = self.state.current_player
|
||||
if mode == "ming" and self.state.hands[player][tile] == 3:
|
||||
self.state.hands[player][tile] -= 3
|
||||
self.state.melds[player].append(("ming_gang", tile))
|
||||
elif mode == "an" and self.state.hands[player][tile] == 4:
|
||||
self.state.hands[player][tile] -= 4
|
||||
self.state.melds[player].append(("an_gang", tile))
|
||||
# 设置缺门：每个玩家定缺（这里假设我们让每个玩家的缺门都为“条”）
|
||||
for player in range(4):
|
||||
missing_suit = "条" # 这里可以通过其他方式设置缺门，比如随机选择
|
||||
self.state.set_missing_suit(player, missing_suit)
|
||||
|
||||
def start_game(self):
|
||||
""" 开始游戏 """
|
||||
if not self.game_started:
|
||||
self.game_started = True
|
||||
logger.info("游戏开始！")
|
||||
else:
|
||||
raise ValueError("杠牌条件不满足")
|
||||
logger.warning("游戏已经开始，不能重复启动！")
|
||||
|
||||
def set_missing_suit(player, missing_suit):
|
||||
# 确定玩家的缺门
|
||||
valid_suits = ["条", "筒", "万"]
|
||||
if missing_suit not in valid_suits:
|
||||
raise ValueError("缺门设置无效")
|
||||
player.missing_suit = missing_suit
|
||||
def check_game_over(self):
|
||||
""" 检查游戏是否结束 """
|
||||
# 你可以根据游戏规则检查是否有玩家胡牌或其他结束条件
|
||||
if len(self.state.deck) == 0:
|
||||
self.game_over = True
|
||||
logger.info("游戏结束！")
|
||||
|
||||
|
||||
def check_blood_battle(self):
|
||||
if len(self.winners) >= 3 or self.state.remaining_tiles == 0:
|
||||
self.game_over = True
|
||||
|
||||
|
||||
def can_win(hand):
|
||||
# 判断是否满足胡牌条件（四组+一对）
|
||||
from collections import Counter
|
||||
|
||||
# 判断缺一门
|
||||
suits = [get_suit(tile) for tile, count in enumerate(hand) if count > 0]
|
||||
if len(set(suits)) > 2:
|
||||
return False # 不满足缺一门
|
||||
|
||||
# 判断是否满足四组+一对
|
||||
def is_valid_group(tiles):
|
||||
# 判断是否为顺子、刻子或对子
|
||||
return len(tiles) == 3 and (tiles[0] == tiles[1] == tiles[2] or
|
||||
tiles[0] + 1 == tiles[1] and tiles[1] + 1 == tiles[2])
|
||||
|
||||
counter = Counter(hand)
|
||||
pairs = [tile for tile, count in counter.items() if count >= 2]
|
||||
for pair in pairs:
|
||||
temp_hand = hand[:]
|
||||
temp_hand[pair] -= 2
|
||||
if is_valid_group(temp_hand):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def calculate_fan(hand, melds, is_self_draw, is_cleared):
|
||||
fan = 1 # 基本胡
|
||||
if is_cleared:
|
||||
fan += 2 # 清一色
|
||||
if len(melds) >= 2:
|
||||
fan += len(melds) # 根据杠加番
|
||||
if is_self_draw:
|
||||
fan += 1 # 自摸加番
|
||||
return fan
|
||||
|
|
|
|||
|
|
@ -1,6 +1,10 @@
|
|||
from .utils import get_suit,get_tile_name
|
||||
from loguru import logger
|
||||
|
||||
|
||||
class ChengduMahjongState:
|
||||
def __init__(self):
|
||||
# 每个玩家的手牌
|
||||
# 每个玩家的手牌，使用108个索引表示
|
||||
self.hands = [[0] * 108 for _ in range(4)] # 每个玩家108张牌的计数
|
||||
# 每个玩家的打出的牌
|
||||
self.discards = [[] for _ in range(4)] # 每个玩家的弃牌列表
|
||||
|
|
@ -10,7 +14,76 @@ class ChengduMahjongState:
|
|||
self.deck = list(range(108)) # 0-107 表示108张牌
|
||||
# 当前玩家索引
|
||||
self.current_player = 0
|
||||
# 玩家分数
|
||||
self.scores = [100, 100, 100, 100]
|
||||
# 剩余牌数量
|
||||
self.remaining_tiles = 108
|
||||
# 胜利玩家列表
|
||||
self.winners = []
|
||||
# 缺门信息
|
||||
self.missing_suits = [None] * 4 # 每个玩家的缺门（"条"、"筒" 或 "万"）
|
||||
|
||||
def set_missing_suit(self, player, missing_suit):
|
||||
"""
|
||||
设置玩家的缺门信息。
|
||||
|
||||
参数:
|
||||
- player: 玩家索引（0-3）。
|
||||
- missing_suit: 玩家选择的缺门（"条"、"筒" 或 "万"）。
|
||||
|
||||
异常:
|
||||
- ValueError: 如果缺门设置无效。
|
||||
"""
|
||||
valid_suits = ["条", "筒", "万"]
|
||||
if missing_suit not in valid_suits:
|
||||
raise ValueError("缺门设置无效")
|
||||
self.missing_suits[player] = missing_suit
|
||||
|
||||
def can_win(self, hand):
|
||||
"""
|
||||
判断是否满足胡牌条件：四组（顺子或刻子）+ 一对将。
|
||||
"""
|
||||
from collections import Counter
|
||||
|
||||
def is_valid_group(tiles):
|
||||
"""
|
||||
判断是否为合法的顺子或刻子。
|
||||
"""
|
||||
if len(tiles) != 3:
|
||||
return False
|
||||
tiles.sort() # 确保顺子检查按顺序排列
|
||||
return (tiles[0] == tiles[1] == tiles[2]) or \
|
||||
(tiles[0] + 1 == tiles[1] and tiles[1] + 1 == tiles[2])
|
||||
|
||||
def try_win(remaining_tiles, depth=0):
|
||||
"""
|
||||
递归检查是否可以将剩余牌分为合法组合。
|
||||
"""
|
||||
if not remaining_tiles:
|
||||
return depth == 4 # 必须分成四组
|
||||
|
||||
for i in range(len(remaining_tiles)):
|
||||
for j in range(i + 1, len(remaining_tiles)):
|
||||
for k in range(j + 1, len(remaining_tiles)):
|
||||
group = [remaining_tiles[i], remaining_tiles[j], remaining_tiles[k]]
|
||||
if is_valid_group(group):
|
||||
next_tiles = remaining_tiles[:i] + remaining_tiles[i + 1:j] + \
|
||||
remaining_tiles[j + 1:k] + remaining_tiles[k + 1:]
|
||||
# 确保顺子检查按顺序排列
|
||||
next_tiles.sort()
|
||||
if try_win(next_tiles, depth + 1):
|
||||
return True
|
||||
return False
|
||||
|
||||
counter = Counter({tile: count for tile, count in enumerate(hand) if count > 0})
|
||||
pairs = [tile for tile, count in counter.items() if count >= 2]
|
||||
|
||||
for pair in pairs:
|
||||
temp_hand = hand[:]
|
||||
temp_hand[pair] -= 2 # 移除将牌
|
||||
remaining_tiles = [tile for tile, count in enumerate(temp_hand) for _ in range(count)]
|
||||
remaining_tiles.sort() # 确保顺子检查按顺序排列
|
||||
if try_win(remaining_tiles):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
|
|
|||
|
|
@ -0,0 +1,39 @@
|
|||
from collections import defaultdict
|
||||
|
||||
from collections import defaultdict
|
||||
|
||||
class Hand:
|
||||
def __init__(self):
|
||||
# 存储所有的牌
|
||||
self.tiles = []
|
||||
# 存储每种牌的数量，默认值为 0
|
||||
self.tile_count = defaultdict(int)
|
||||
|
||||
def add_tile(self, tile):
|
||||
""" 向手牌中添加一张牌 """
|
||||
self.tiles.append(tile) # 将牌添加到手牌中
|
||||
self.tile_count[tile] += 1 # 增加牌的数量
|
||||
|
||||
def remove_tile(self, tile):
|
||||
""" 从手牌中移除一张牌 """
|
||||
if self.tile_count[tile] > 0:
|
||||
self.tiles.remove(tile)
|
||||
self.tile_count[tile] -= 1
|
||||
else:
|
||||
raise ValueError(f"手牌中没有该牌: {tile}")
|
||||
|
||||
def get_tile_count(self, tile):
|
||||
""" 获取手牌中某张牌的数量 """
|
||||
return self.tile_count[tile]
|
||||
|
||||
def can_peng(self, tile):
|
||||
""" 判断是否可以碰（即是否已经有2张相同的牌，摸一张牌后可以碰） """
|
||||
return self.tile_count[tile] == 2 # 摸一张牌后总数为 3 张，才可以碰
|
||||
|
||||
def can_gang(self, tile):
|
||||
""" 判断是否可以杠（即是否已经有3张相同的牌，摸一张牌后可以杠） """
|
||||
return self.tile_count[tile] == 3 # 摸一张牌后总数为 4 张，才可以杠
|
||||
|
||||
def __repr__(self):
|
||||
""" 返回手牌的字符串表示 """
|
||||
return f"手牌: {self.tiles}, 牌的数量: {dict(self.tile_count)}"
|
||||
|
|
@ -0,0 +1,20 @@
|
|||
class MahjongTile:
|
||||
SUITS = ['条', '筒', '万']
|
||||
|
||||
def __init__(self, suit, value):
|
||||
if suit not in self.SUITS or not (1 <= value <= 9):
|
||||
raise ValueError("Invalid tile")
|
||||
self.suit = suit
|
||||
self.value = value
|
||||
|
||||
def __repr__(self):
|
||||
return f"{self.value}{self.suit}"
|
||||
|
||||
def __eq__(self, other):
|
||||
return self.suit == other.suit and self.value == other.value
|
||||
|
||||
def __hash__(self):
|
||||
return hash((self.suit, self.value))
|
||||
|
||||
|
||||
|
||||
|
|
@ -1,13 +1,32 @@
|
|||
def can_win(hand):
|
||||
# 判断是否满足胡牌条件
|
||||
...
|
||||
def calculate_score(fan: int, base_score: int, is_self_draw: bool) -> dict:
|
||||
"""
|
||||
根据成都麻将规则计算得分（不区分庄家和闲家）。
|
||||
|
||||
参数:
|
||||
- fan: 总番数。
|
||||
- base_score: 底分。
|
||||
- is_self_draw: 是否为自摸。
|
||||
|
||||
返回:
|
||||
- scores: 字典，包含赢家得分和输家扣分。
|
||||
"""
|
||||
# 计算总分
|
||||
multiplier = 2 ** fan # 根据番数计算倍率
|
||||
total_score = base_score * multiplier
|
||||
|
||||
def calculate_fan(hand, melds, is_self_draw, is_cleared):
|
||||
fan = 1 # 基本胡
|
||||
if is_cleared:
|
||||
fan += 2 # 清一色
|
||||
if len(melds) >= 2:
|
||||
fan += len(melds)
|
||||
if is_self_draw:
|
||||
fan += 1
|
||||
return fan
|
||||
# 自摸：三家平摊分数
|
||||
per_loser_score = -total_score
|
||||
winner_score = total_score * 3 # 总赢家得分
|
||||
return {
|
||||
"winner": winner_score,
|
||||
"loser": [per_loser_score] * 3
|
||||
}
|
||||
else:
|
||||
# 点炮：点炮者独付
|
||||
loser_score = -total_score
|
||||
winner_score = total_score
|
||||
return {
|
||||
"winner": winner_score,
|
||||
"loser": [loser_score, 0, 0]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,19 @@
|
|||
def get_suit(tile_index):
|
||||
"""
|
||||
根据牌的索引返回花色。
|
||||
条：索引 0-35，筒：索引 36-71，万：索引 72-107
|
||||
"""
|
||||
suits = ["条", "筒", "万"]
|
||||
return suits[tile_index // 36]
|
||||
if 0 <= tile_index <= 35:
|
||||
return "条"
|
||||
elif 36 <= tile_index <= 71:
|
||||
return "筒"
|
||||
elif 72 <= tile_index <= 107:
|
||||
return "万"
|
||||
else:
|
||||
raise ValueError(f"无效的牌索引: {tile_index}")
|
||||
|
||||
def get_tile_name(tile_index):
|
||||
"""
|
||||
根据牌的索引返回牌名（例如：1条，2筒等）。
|
||||
"""
|
||||
suit = get_suit(tile_index)
|
||||
return f"{tile_index % 36 + 1}{suit}"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,120 @@
|
|||
import gym
|
||||
import numpy as np
|
||||
from gym import spaces
|
||||
from src.engine.actions import draw_tile, discard_tile, peng, gang, check_blood_battle
|
||||
from src.engine.calculate_fan import calculate_fan, is_seven_pairs, is_cleared, is_big_pairs
|
||||
from src.engine.chengdu_mahjong_engine import ChengduMahjongEngine
|
||||
from src.engine.scoring import calculate_score
|
||||
|
||||
|
||||
class MahjongEnv(gym.Env):
|
||||
def __init__(self):
|
||||
super(MahjongEnv, self).__init__()
|
||||
self.engine = ChengduMahjongEngine()
|
||||
self.scores = [100, 100, 100, 100] # 四位玩家初始分数
|
||||
self.base_score = 1 # 底分
|
||||
self.max_rounds = 100 # 最大轮数，防止游戏无限进行
|
||||
self.current_round = 0 # 当前轮数
|
||||
self.action_space = spaces.Discrete(108) # 动作空间：打牌的索引
|
||||
self.observation_space = spaces.Box(low=0, high=4, shape=(108,), dtype=np.int32)
|
||||
|
||||
def reset(self):
|
||||
self.engine = ChengduMahjongEngine()
|
||||
self.scores = [100, 100, 100, 100] # 每局重置分数
|
||||
self.current_round = 0
|
||||
return self.engine.state.hands[self.engine.state.current_player]
|
||||
|
||||
def step(self, action):
|
||||
"""
|
||||
执行玩家动作并更新游戏状态。
|
||||
|
||||
参数:
|
||||
- action: 玩家动作，0 代表摸牌，1 代表打牌，2 代表碰牌，3 代表杠牌
|
||||
|
||||
返回:
|
||||
- next_state: 当前玩家的手牌
|
||||
- reward: 奖励
|
||||
- done: 是否结束
|
||||
- info: 其他信息（如奖励历史等）
|
||||
"""
|
||||
done = False
|
||||
reward = 0
|
||||
|
||||
try:
|
||||
if action == 0: # 0代表摸牌
|
||||
reward, done = draw_tile(self.engine) # 调用摸牌函数
|
||||
elif action == 1: # 1代表打牌
|
||||
tile = self.engine.state.hands[self.engine.state.current_player][0] # 假设选择第一张牌
|
||||
discard_tile(self.engine, tile) # 调用打牌函数
|
||||
reward, done = -1, False
|
||||
elif action == 2: # 2代表碰牌
|
||||
tile = self.engine.state.hands[self.engine.state.current_player][0] # 假设选择第一张牌
|
||||
peng(self.engine, tile) # 调用碰牌函数
|
||||
reward, done = 0, False
|
||||
elif action == 3: # 3代表杠牌
|
||||
tile = self.engine.state.hands[self.engine.state.current_player][0] # 假设选择第一张牌
|
||||
gang(self.engine, tile, mode="ming") # 暂时假设为明杠
|
||||
reward, done = 0, False
|
||||
|
||||
# 检查是否胡牌
|
||||
if self.engine.state.can_win(self.engine.state.hands[self.engine.state.current_player]):
|
||||
reward, done = self.handle_win() # 胡牌时处理胜利逻辑
|
||||
|
||||
# 检查游戏结束条件
|
||||
check_blood_battle(self.engine)
|
||||
|
||||
if self.engine.game_over: # 检查是否游戏结束
|
||||
done = True
|
||||
|
||||
except ValueError:
|
||||
reward, done = -10, False # 非法操作扣分
|
||||
|
||||
# 切换到下一个玩家
|
||||
self.engine.state.current_player = (self.engine.state.current_player + 1) % 4
|
||||
self.current_round += 1
|
||||
|
||||
# 如果达到最大轮数，结束游戏
|
||||
if self.current_round >= self.max_rounds:
|
||||
done = True
|
||||
reward = 0 # 平局奖励或惩罚（可调整）
|
||||
|
||||
return self.engine.state.hands[self.engine.state.current_player], reward, done, {}
|
||||
|
||||
def handle_win(self):
|
||||
"""
|
||||
处理胡牌后的分数结算和奖励。
|
||||
"""
|
||||
winner = self.engine.state.current_player
|
||||
hand = self.engine.state.hands[winner]
|
||||
melds = self.engine.state.melds[winner]
|
||||
is_self_draw = True # 假设自摸（后续可动态判断）
|
||||
|
||||
conditions = {
|
||||
"is_cleared": is_cleared(hand, melds),
|
||||
"is_seven_pairs": is_seven_pairs(hand),
|
||||
"is_big_pairs": is_big_pairs(hand),
|
||||
# 添加其他条件...
|
||||
}
|
||||
|
||||
# 动态计算番数
|
||||
fan = calculate_fan(hand, melds, is_self_draw, is_cleared, conditions)
|
||||
|
||||
# 动态计算得分
|
||||
scores = calculate_score(fan, self.base_score, is_self_draw)
|
||||
self.scores[winner] += scores["winner"]
|
||||
for i, score in enumerate(scores["loser"]):
|
||||
self.scores[i] += score # 扣分
|
||||
|
||||
# 奖励设置为赢家得分
|
||||
reward = scores["winner"]
|
||||
self.engine.state.winners.append(winner) # 添加赢家到列表
|
||||
return reward, True # 胡牌结束当前局
|
||||
|
||||
def render(self, mode="human"):
|
||||
"""
|
||||
打印游戏状态信息，便于调试。
|
||||
"""
|
||||
print(f"当前轮数: {self.current_round}")
|
||||
print("玩家分数:", self.scores)
|
||||
print("当前玩家状态:", self.engine.state.hands[self.engine.state.current_player])
|
||||
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
import torch
|
||||
print(torch.cuda.is_available()) # 如果返回True，说明可以使用GPU
|
||||
print(torch.__version__)
|
||||
|
|
@ -0,0 +1,267 @@
|
|||
import pytest
|
||||
from src.engine.calculate_fan import calculate_fan, is_seven_pairs, is_cleared, is_big_pairs
|
||||
|
||||
# 测试用例
|
||||
|
||||
def test_basic_win():
|
||||
"""
|
||||
测试平胡（基本胡）计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
# 模拟平胡手牌: 四组顺子 + 一对将
|
||||
hand[0] = 2 # 将: 两张1条
|
||||
hand[3] = 1 # 2条
|
||||
hand[4] = 1 # 3条
|
||||
hand[5] = 1 # 4条
|
||||
hand[10] = 1 # 5条
|
||||
hand[11] = 1 # 6条
|
||||
hand[12] = 1 # 7条
|
||||
hand[20] = 1 # 8条
|
||||
hand[21] = 1 # 9条
|
||||
hand[22] = 1 # 1筒
|
||||
hand[30] = 1 # 2筒
|
||||
hand[31] = 1 # 3筒
|
||||
|
||||
melds = []
|
||||
conditions = {}
|
||||
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=False, conditions=conditions)
|
||||
assert fan == 1, f"Expected 1 fan, got {fan}"
|
||||
|
||||
|
||||
def test_clear_win():
|
||||
"""
|
||||
测试清一色计分（不加自摸番）
|
||||
"""
|
||||
hand = [0] * 108
|
||||
# 模拟清一色手牌
|
||||
hand[0] = 2
|
||||
hand[4] = 1
|
||||
hand[5] = 1
|
||||
hand[6] = 1
|
||||
hand[10] = 1
|
||||
hand[11] = 1
|
||||
hand[12] = 1
|
||||
hand[20] = 1
|
||||
hand[21] = 1
|
||||
hand[22] = 1
|
||||
hand[23] = 1
|
||||
|
||||
melds = []
|
||||
conditions = {"is_seven_pairs": False, "add_self_draw": False}
|
||||
|
||||
fan = calculate_fan(hand, melds, is_self_draw=True, is_cleared=is_cleared(hand), conditions=conditions)
|
||||
assert fan == 3, f"Expected 3 fans (1 basic + 2 cleared), got {fan}"
|
||||
|
||||
def test_pure_cleared():
|
||||
"""
|
||||
测试清对计分
|
||||
"""
|
||||
hand = [2] * 6 + [0] * 102 # 手牌：清对
|
||||
melds = [0, 1, 2] # 模拟碰
|
||||
conditions = {"is_pure_cleared": True}
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=True, conditions=conditions)
|
||||
assert fan == 3, f"Expected 3 fans (pure cleared), got {fan}"
|
||||
|
||||
|
||||
|
||||
def test_pure_cleared():
|
||||
"""
|
||||
测试清对计分
|
||||
"""
|
||||
hand = [2] * 6 + [0] * 102 # 手牌：清对
|
||||
melds = [0, 1, 2] # 模拟碰
|
||||
conditions = {"is_pure_cleared": True} # 明确条件
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=True, conditions=conditions)
|
||||
assert fan == 3, f"Expected 3 fans (pure cleared), got {fan}"
|
||||
|
||||
def test_seven_pairs():
|
||||
"""
|
||||
测试七对计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
# 模拟七对手牌
|
||||
hand[0] = 2 # 1条
|
||||
hand[4] = 2 # 2条
|
||||
hand[8] = 2 # 3条
|
||||
hand[12] = 2 # 4条
|
||||
hand[16] = 2 # 5条
|
||||
hand[20] = 2 # 6条
|
||||
hand[24] = 2 # 7条
|
||||
|
||||
melds = []
|
||||
conditions = {"is_seven_pairs": is_seven_pairs(hand)}
|
||||
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=False, conditions=conditions)
|
||||
assert fan == 2, f"Expected 2 fans (7 pairs), got {fan}"
|
||||
|
||||
|
||||
def test_small_pairs():
|
||||
"""
|
||||
测试小七对计分
|
||||
"""
|
||||
hand = [2] * 6 + [0] * 102
|
||||
melds = []
|
||||
conditions = {"is_small_pairs": True}
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=False, conditions=conditions)
|
||||
assert fan == 2, f"Expected 2 fans (small pairs), got {fan}"
|
||||
|
||||
|
||||
def test_clear_seven_pairs():
|
||||
"""
|
||||
测试清七对计分
|
||||
"""
|
||||
hand = [2] * 7 + [0] * 101
|
||||
melds = []
|
||||
conditions = {"is_clear_seven_pairs": True}
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=True, conditions=conditions)
|
||||
assert fan == 12, f"Expected 12 fans (clear seven pairs), got {fan}"
|
||||
|
||||
|
||||
|
||||
def test_full_request():
|
||||
"""
|
||||
测试全求人计分
|
||||
"""
|
||||
hand = [2] + [0] * 107
|
||||
melds = []
|
||||
conditions = {"is_full_request": True}
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=False, conditions=conditions)
|
||||
assert fan == 6, f"Expected 6 fans (full request), got {fan}"
|
||||
|
||||
|
||||
def test_big_pairs():
|
||||
"""
|
||||
测试大对子计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
# 模拟大对子手牌
|
||||
hand[0] = 3 # 1条
|
||||
hand[4] = 3 # 2条
|
||||
hand[8] = 3 # 3条
|
||||
hand[12] = 2 # 将: 4条
|
||||
|
||||
melds = []
|
||||
conditions = {"is_big_pairs": is_big_pairs(hand)}
|
||||
|
||||
# Debug output
|
||||
print(f"Conditions: {conditions}")
|
||||
|
||||
# 确保大对子检测正确
|
||||
assert is_big_pairs(hand), "The hand is not identified as a big pairs hand."
|
||||
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=False, conditions=conditions)
|
||||
assert fan == 2, f"Expected 2 fans (big pairs), got {fan}"
|
||||
|
||||
|
||||
|
||||
def test_gang_flower():
|
||||
"""
|
||||
测试杠上开花计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
# 模拟杠上开花手牌
|
||||
hand[0] = 2 # 将: 两张1条
|
||||
hand[3] = 1 # 2条
|
||||
hand[4] = 1 # 3条
|
||||
hand[5] = 1 # 4条
|
||||
hand[10] = 1 # 5条
|
||||
hand[11] = 1 # 6条
|
||||
hand[12] = 1 # 7条
|
||||
hand[20] = 1 # 8条
|
||||
hand[21] = 1 # 9条
|
||||
hand[22] = 1 # 1筒
|
||||
|
||||
melds = []
|
||||
conditions = {"is_gang_flower": True}
|
||||
|
||||
fan = calculate_fan(hand, melds, is_self_draw=True, is_cleared=False, conditions=conditions)
|
||||
assert fan == 2, f"Expected 2 fans (1 basic + 1 gang flower), got {fan}"
|
||||
|
||||
|
||||
def test_rob_gang():
|
||||
"""
|
||||
测试抢杠胡计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
melds = []
|
||||
conditions = {"is_rob_gang": True}
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=False, conditions=conditions)
|
||||
assert fan == 1, f"Expected 1 fan (rob gang), got {fan}"
|
||||
|
||||
def test_under_the_sea():
|
||||
"""
|
||||
测试海底捞月计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
melds = []
|
||||
conditions = {"is_under_the_sea": True}
|
||||
fan = calculate_fan(hand, melds, is_self_draw=True, is_cleared=False, conditions=conditions)
|
||||
assert fan == 1, f"Expected 1 fan (under the sea), got {fan}"
|
||||
|
||||
|
||||
|
||||
def test_cannon():
|
||||
"""
|
||||
测试放炮计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
melds = []
|
||||
conditions = {"is_cannon": True}
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=False, conditions=conditions)
|
||||
assert fan == 1, f"Expected 1 fan (cannon), got {fan}"
|
||||
|
||||
|
||||
def test_tian_hu():
|
||||
"""
|
||||
测试天胡计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
melds = []
|
||||
conditions = {"is_tian_hu": True}
|
||||
fan = calculate_fan(hand, melds, is_self_draw=True, is_cleared=False, conditions=conditions)
|
||||
assert fan == 12, f"Expected 12 fans (tian hu), got {fan}"
|
||||
|
||||
def test_di_hu():
|
||||
"""
|
||||
测试地胡计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
melds = []
|
||||
conditions = {"is_di_hu": True}
|
||||
fan = calculate_fan(hand, melds, is_self_draw=False, is_cleared=False, conditions=conditions)
|
||||
assert fan == 12, f"Expected 12 fans (di hu), got {fan}"
|
||||
|
||||
def test_dragon_seven_pairs():
|
||||
"""
|
||||
测试龙七对计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
# 模拟龙七对手牌
|
||||
hand[0] = 2 # 1条
|
||||
hand[4] = 2 # 2条
|
||||
hand[8] = 2 # 3条
|
||||
hand[12] = 2 # 4条
|
||||
hand[16] = 2 # 5条
|
||||
hand[20] = 2 # 6条
|
||||
hand[24] = 4 # 龙: 7条
|
||||
|
||||
melds = []
|
||||
conditions = {
|
||||
"is_dragon_seven_pairs": True,
|
||||
"is_seven_pairs": True,
|
||||
}
|
||||
|
||||
fan = calculate_fan(hand, melds, is_self_draw=True, is_cleared=False, conditions=conditions)
|
||||
assert fan == 12, f"Expected 13 fans (1 self-draw + 12 dragon seven pairs), got {fan}"
|
||||
|
||||
def test_self_draw():
|
||||
"""
|
||||
测试自摸计分
|
||||
"""
|
||||
hand = [0] * 108
|
||||
melds = []
|
||||
conditions = {}
|
||||
fan = calculate_fan(hand, melds, is_self_draw=True, is_cleared=False, conditions=conditions)
|
||||
assert fan == 1, f"Expected 1 fan (self-draw), got {fan}"
|
||||
|
||||
|
|
@ -0,0 +1,96 @@
|
|||
def test_draw_tile():
|
||||
from src.engine.chengdu_mahjong_engine import ChengduMahjongEngine
|
||||
|
||||
engine = ChengduMahjongEngine()
|
||||
initial_remaining = engine.state.remaining_tiles
|
||||
tile = engine.draw_tile()
|
||||
|
||||
# 验证牌堆数量减少
|
||||
assert engine.state.remaining_tiles == initial_remaining - 1, "牌堆数量未正确减少"
|
||||
# 验证牌已加入当前玩家手牌
|
||||
assert engine.state.hands[engine.state.current_player][tile] > 0, "摸牌未加入玩家手牌"
|
||||
print(f"test_draw_tile passed: 摸到了 {tile}")
|
||||
|
||||
|
||||
def test_discard_tile():
|
||||
from src.engine.chengdu_mahjong_engine import ChengduMahjongEngine
|
||||
|
||||
engine = ChengduMahjongEngine()
|
||||
tile = engine.draw_tile() # 玩家先摸牌
|
||||
engine.discard_tile(tile) # 打出摸到的牌
|
||||
|
||||
# 验证手牌数量减少
|
||||
assert engine.state.hands[engine.state.current_player][tile] == 0, "手牌未正确移除"
|
||||
# 验证牌加入了牌河
|
||||
assert tile in engine.state.discards[engine.state.current_player], "牌未正确加入牌河"
|
||||
print(f"test_discard_tile passed: 打出了 {tile}")
|
||||
|
||||
|
||||
def test_set_missing_suit():
|
||||
from src.engine.game_state import ChengduMahjongState
|
||||
|
||||
state = ChengduMahjongState()
|
||||
player = 0
|
||||
missing_suit = "筒"
|
||||
|
||||
state.set_missing_suit(player, missing_suit)
|
||||
|
||||
# 验证缺门是否正确设置
|
||||
assert state.missing_suits[player] == missing_suit, "缺门设置错误"
|
||||
print(f"test_set_missing_suit passed: 缺门设置为 {missing_suit}")
|
||||
|
||||
|
||||
def test_can_win():
|
||||
from src.engine.game_state import ChengduMahjongState
|
||||
|
||||
state = ChengduMahjongState()
|
||||
hand = [0] * 108
|
||||
hand[0] = 2 # 两张1条（对子）
|
||||
hand[3] = 1 # 2条
|
||||
hand[4] = 1 # 3条
|
||||
hand[5] = 1 # 4条
|
||||
hand[10] = 1 # 5条
|
||||
hand[11] = 1 # 6条
|
||||
hand[12] = 1 # 7条
|
||||
hand[20] = 1 # 8条
|
||||
hand[21] = 1 # 9条
|
||||
hand[22] = 1 # 1筒
|
||||
hand[30] = 1 # 2筒
|
||||
hand[31] = 1 # 3筒
|
||||
hand[32] = 1 # 4筒
|
||||
|
||||
result = state.can_win(hand)
|
||||
|
||||
assert result is True, "胡牌判断失败"
|
||||
print(f"test_can_win passed: 胡牌条件正确")
|
||||
|
||||
|
||||
|
||||
|
||||
def test_peng():
|
||||
from src.engine.chengdu_mahjong_engine import ChengduMahjongEngine
|
||||
|
||||
engine = ChengduMahjongEngine()
|
||||
tile = 5 # 模拟手牌中有3张牌
|
||||
engine.state.hands[engine.state.current_player][tile] = 3
|
||||
engine.peng(tile)
|
||||
|
||||
# 验证手牌减少
|
||||
assert engine.state.hands[engine.state.current_player][tile] == 1, "碰牌后手牌数量错误"
|
||||
# 验证明牌记录
|
||||
assert ("peng", tile) in engine.state.melds[engine.state.current_player], "碰牌未正确记录"
|
||||
print(f"test_peng passed: 碰牌成功")
|
||||
|
||||
def test_gang():
|
||||
from src.engine.chengdu_mahjong_engine import ChengduMahjongEngine
|
||||
|
||||
engine = ChengduMahjongEngine()
|
||||
tile = 10 # 模拟手牌中有4张牌
|
||||
engine.state.hands[engine.state.current_player][tile] = 4
|
||||
engine.gang(tile, mode="an")
|
||||
|
||||
# 验证手牌减少
|
||||
assert engine.state.hands[engine.state.current_player][tile] == 0, "杠牌后手牌数量错误"
|
||||
# 验证明牌记录
|
||||
assert ("an_gang", tile) in engine.state.melds[engine.state.current_player], "杠牌未正确记录"
|
||||
print(f"test_gang passed: 杠牌成功")
|
||||
|
|
@ -0,0 +1,64 @@
|
|||
from src.engine.hand import Hand
|
||||
|
||||
|
||||
def test_hand():
|
||||
# 创建一个玩家的手牌
|
||||
hand = Hand()
|
||||
|
||||
# 添加一些牌到手牌中
|
||||
hand.add_tile("1条")
|
||||
hand.add_tile("1条")
|
||||
hand.add_tile("2条")
|
||||
hand.add_tile("2条")
|
||||
hand.add_tile("2条")
|
||||
hand.add_tile("3条")
|
||||
|
||||
# 打印手牌
|
||||
print("\n当前手牌:", hand)
|
||||
|
||||
# 测试获取某张牌的数量
|
||||
assert hand.get_tile_count("1条") == 2, f"测试失败：1条应该有 2 张"
|
||||
assert hand.get_tile_count("2条") == 3, f"测试失败：2条应该有 3 张"
|
||||
assert hand.get_tile_count("3条") == 1, f"测试失败：3条应该有 1 张"
|
||||
|
||||
# 测试移除一张牌
|
||||
hand.remove_tile("1条")
|
||||
print("移除 1条 后的手牌:", hand)
|
||||
assert hand.get_tile_count("1条") == 1, f"测试失败：1条应该有 1 张"
|
||||
|
||||
# 确保移除后有足够的牌可以碰
|
||||
# 添加一张 1条，确保可以碰
|
||||
hand.add_tile("1条")
|
||||
print("添加 1条 后的手牌:", hand)
|
||||
|
||||
# 测试是否可以碰
|
||||
assert hand.can_peng("1条") == True, f"测试失败：1条应该可以碰"
|
||||
print("可以碰 1条 的牌:", hand.can_peng("1条"))
|
||||
assert hand.can_peng("3条") == False, f"测试失败：3条不可以碰"
|
||||
print("不可以碰 3条 的牌:", hand.can_peng("3条"))
|
||||
|
||||
# 测试是否可以杠
|
||||
assert hand.can_gang("1条") == False, f"测试失败：1条不可以杠"
|
||||
print("不可以杠 1条 的牌:", hand.can_gang("1条"))
|
||||
assert hand.can_gang("2条") == False, f"测试失败：2条不可以杠"
|
||||
print("不可以杠 2条 的牌:", hand.can_gang("2条"))
|
||||
|
||||
# 添加更多牌来形成杠
|
||||
hand.add_tile("2条")
|
||||
print("添加牌后手牌:", hand)
|
||||
hand.add_tile("2条")
|
||||
print("添加牌后手牌:", hand)
|
||||
assert hand.can_gang("2条") == False, f"测试失败：2条不可以杠" # still not enough for gang
|
||||
|
||||
# 添加一张更多的 2条 来形成杠
|
||||
hand.add_tile("2条")
|
||||
print("添加一张2条后:", hand)
|
||||
assert hand.can_gang("2条") == True, f"测试失败：2条应该可以杠"
|
||||
|
||||
print("所有测试通过！")
|
||||
|
||||
# 运行测试
|
||||
test_hand()
|
||||
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,45 @@
|
|||
from src.engine.mahjong_tile import MahjongTile
|
||||
|
||||
def test_mahjong_tile():
|
||||
# 测试合法的牌
|
||||
tile1 = MahjongTile("条", 5)
|
||||
assert tile1.suit == "条", f"测试失败：预期花色是 '条'，但实际是 {tile1.suit}"
|
||||
assert tile1.value == 5, f"测试失败：预期面值是 5，但实际是 {tile1.value}"
|
||||
assert repr(tile1) == "5条", f"测试失败：预期牌名是 '5条'，但实际是 {repr(tile1)}"
|
||||
|
||||
tile2 = MahjongTile("筒", 3)
|
||||
assert tile2.suit == "筒", f"测试失败：预期花色是 '筒'，但实际是 {tile2.suit}"
|
||||
assert tile2.value == 3, f"测试失败：预期面值是 3，但实际是 {tile2.value}"
|
||||
assert repr(tile2) == "3筒", f"测试失败：预期牌名是 '3筒'，但实际是 {repr(tile2)}"
|
||||
|
||||
tile3 = MahjongTile("万", 9)
|
||||
assert tile3.suit == "万", f"测试失败：预期花色是 '万'，但实际是 {tile3.suit}"
|
||||
assert tile3.value == 9, f"测试失败：预期面值是 9，但实际是 {tile3.value}"
|
||||
assert repr(tile3) == "9万", f"测试失败：预期牌名是 '9万'，但实际是 {repr(tile3)}"
|
||||
|
||||
# 测试非法的牌
|
||||
try:
|
||||
MahjongTile("条", 10) # 面值超出范围
|
||||
assert False, "测试失败：面值为 10 的牌应该抛出异常"
|
||||
except ValueError:
|
||||
pass # 正确抛出异常
|
||||
|
||||
try:
|
||||
MahjongTile("花", 5) # 花色无效
|
||||
assert False, "测试失败：花色为 '花' 的牌应该抛出异常"
|
||||
except ValueError:
|
||||
pass # 正确抛出异常
|
||||
|
||||
# 测试相等判断
|
||||
tile4 = MahjongTile("条", 5)
|
||||
assert tile1 == tile4, f"测试失败：预期 {tile1} 和 {tile4} 相等"
|
||||
tile5 = MahjongTile("筒", 5)
|
||||
assert tile1 != tile5, f"测试失败：预期 {tile1} 和 {tile5} 不相等"
|
||||
|
||||
# 测试哈希
|
||||
tile_set = {tile1, tile4, tile2}
|
||||
assert len(tile_set) == 2, f"测试失败：集合中应该有 2 张牌，而实际有 {len(tile_set)} 张"
|
||||
|
||||
print("所有测试通过！")
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
import pytest
|
||||
from src.engine.scoring import calculate_score
|
||||
|
||||
@pytest.mark.parametrize("fan, is_self_draw, base_score, expected_scores", [
|
||||
# 测试用例 1: 自摸，总番数 3
|
||||
(3, True, 5, {"winner": 120, "loser": [-40, -40, -40]}),
|
||||
|
||||
# 测试用例 2: 点炮，总番数 2
|
||||
(2, False, 5, {"winner": 20, "loser": [-20, 0, 0]}),
|
||||
|
||||
# 测试用例 3: 自摸，总番数 4
|
||||
(4, True, 5, {"winner": 240, "loser": [-80, -80, -80]}),
|
||||
|
||||
# 测试用例 4: 点炮，总番数 1
|
||||
(1, False, 5, {"winner": 10, "loser": [-10, 0, 0]}),
|
||||
])
|
||||
def test_calculate_score(fan, is_self_draw, base_score, expected_scores):
|
||||
scores = calculate_score(fan, base_score, is_self_draw)
|
||||
assert scores == expected_scores, f"测试失败: {scores} != {expected_scores}"
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
from src.engine.utils import get_suit,get_tile_name
|
||||
|
||||
def test_get_suit():
|
||||
# 测试条花色（0-35）
|
||||
for i in range(36):
|
||||
assert get_suit(i) == "条", f"测试失败：索引 {i} 应该是 '条'"
|
||||
|
||||
# 测试筒花色（36-71）
|
||||
for i in range(36, 72):
|
||||
assert get_suit(i) == "筒", f"测试失败：索引 {i} 应该是 '筒'"
|
||||
|
||||
# 测试万花色（72-107）
|
||||
for i in range(72, 108):
|
||||
assert get_suit(i) == "万", f"测试失败：索引 {i} 应该是 '万'"
|
||||
|
||||
# 测试无效索引
|
||||
try:
|
||||
get_suit(108)
|
||||
assert False, "测试失败：索引 108 应该抛出 ValueError"
|
||||
except ValueError:
|
||||
pass # 如果抛出 ValueError，测试通过
|
||||
|
||||
print("get_suit 测试通过！")
|
||||
|
||||
def test_get_tile_name():
|
||||
# 测试每个牌的名称是否正确
|
||||
for i in range(108):
|
||||
tile_name = get_tile_name(i)
|
||||
assert tile_name == f"{i % 36 + 1}{get_suit(i)}", \
|
||||
f"测试失败：索引 {i} 应该是 '{i % 36 + 1}{get_suit(i)}'，但实际返回 '{tile_name}'"
|
||||
|
||||
print("get_tile_name 测试通过！")
|
||||
|
||||
# 运行测试
|
||||
test_get_suit()
|
||||
test_get_tile_name()
|
||||
Loading…
Reference in New Issue