feat: combat_engine + generate_pnj_batch + actions/encounter updated
This commit is contained in:
Corback
@@ -223,21 +223,21 @@
|
||||
|
||||
"outcomes": {
|
||||
"combat": {
|
||||
"win": {"loot": true, "xp": true, "stress_delta": +2, "description": "Victoire au combat"},
|
||||
"loss": {"loot": false, "xp": false, "stress_delta": +5, "hp_loss_pct": 0.25, "description": "Défaite"},
|
||||
"draw": {"loot": false, "xp": true, "stress_delta": +3, "hp_loss_pct": 0.15, "description": "Combat indécis"}
|
||||
"win": {"loot": true, "xp": true, "stress_delta": 2, "description": "Victoire au combat"},
|
||||
"loss": {"loot": false, "xp": false, "stress_delta": 5, "hp_loss_pct": 0.25, "description": "Défaite"},
|
||||
"draw": {"loot": false, "xp": true, "stress_delta": 3, "hp_loss_pct": 0.15, "description": "Combat indécis"}
|
||||
},
|
||||
"fuite": {
|
||||
"success": {"loot": false, "xp": false, "stress_delta": +1, "description": "Fuite réussie"},
|
||||
"failed": {"loot": false, "xp": false, "stress_delta": +3, "hp_loss_pct": 0.10, "description": "Fuite ratée, touché"}
|
||||
"success": {"loot": false, "xp": false, "stress_delta": 1, "description": "Fuite réussie"},
|
||||
"failed": {"loot": false, "xp": false, "stress_delta": 3, "hp_loss_pct": 0.10, "description": "Fuite ratée, touché"}
|
||||
},
|
||||
"evitement": {
|
||||
"success": {"loot": false, "xp": true, "stress_delta": 0, "description": "Discrétion réussie"},
|
||||
"failed": {"loot": false, "xp": false, "stress_delta": +2, "description": "Repéré, doit fuir"}
|
||||
"failed": {"loot": false, "xp": false, "stress_delta": 2, "description": "Repéré, doit fuir"}
|
||||
},
|
||||
"negociation": {
|
||||
"success": {"loot": false, "xp": true, "stress_delta": +1, "caps_cost": true, "description": "Négociation réussie"},
|
||||
"failed": {"loot": false, "xp": false, "stress_delta": +4, "hp_loss_pct": 0.15, "description": "Négociation échouée, attaqué"}
|
||||
"success": {"loot": false, "xp": true, "stress_delta": 1, "caps_cost": true, "description": "Négociation réussie"},
|
||||
"failed": {"loot": false, "xp": false, "stress_delta": 4, "hp_loss_pct": 0.15, "description": "Négociation échouée, attaqué"}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ Règles (Résumé Écran MJ + Résumé des Règles FR) :
|
||||
import random
|
||||
import db
|
||||
import inventory as inv_module
|
||||
import combat_engine
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Rang de compétence par défaut (sans table skills complète en DB)
|
||||
@@ -437,7 +438,7 @@ def _execute_action(char: dict, decision: dict,
|
||||
target = chars_map.get(decision["target_id"])
|
||||
if target and target.get("is_alive", True):
|
||||
weapon = _get_equipped_weapon(inv)
|
||||
resolve_attack(char, target, weapon, session_id, day, tick)
|
||||
combat_engine.resolve_attack(char, target, weapon, session_id, day, tick)
|
||||
|
||||
elif action == "eat":
|
||||
food = next((i for i in inv if i["item_type"] == "food" and i["quantity"] >= 1), None)
|
||||
|
||||
@@ -0,0 +1,393 @@
|
||||
"""
|
||||
Combat Engine — Fallout : Venice of Wasteland
|
||||
Résolution complète d'un round de combat 2D20.
|
||||
|
||||
Règles (Résumé Écran MJ + Résumé des Règles FR) :
|
||||
Test attaque : 2d20 ≤ Attribut + rang_compétence (1=crit=2 réussites, 20=complication)
|
||||
Dés de combat : d6 — 1→1dmg, 2→2dmg, 3-4→0, 5-6→1dmg+1effet
|
||||
Corps à corps : FOR + Corps à corps | Bonus melee : FOR<7→+0, 7-8→+1CD, 9-10→+2CD, 11+→+3CD
|
||||
Distance légère: AGI + Arme légère | Distance lourde: END + Armes lourdes
|
||||
Énergie : PER + Armes à énergie
|
||||
Munitions : 1 de base (distance), +1 par effet "En rafale"
|
||||
Couverture : réduit les dégâts (Résistance = n CD lancés, on soustrait le résultat)
|
||||
Blessures crit : ≥5 dégâts après RD en un seul coup → jet d20 localisation
|
||||
Effets d'armes : Brutal, Perforant X, De zone, Étourdissant, Persistant, Radioactif, En rafale
|
||||
"""
|
||||
|
||||
import random
|
||||
import db
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Table de localisation des blessures critiques (Résumé Écran MJ)
|
||||
# Déclenchée si dégâts ≥ 5 après couverture en un seul coup
|
||||
# ---------------------------------------------------------------------------
|
||||
_CRIT_LOCATION_TABLE = [
|
||||
(2, "bras", "Laisse tomber l'objet porté. Plus d'action avec ce bras ce tour."),
|
||||
(8, "bras", "Laisse tomber l'objet porté. Plus d'action avec ce bras ce tour."),
|
||||
(11, "jambe", "Tombe à terre. Pas de sprint. Le déplacement devient une action capitale."),
|
||||
(14, "jambe", "Tombe à terre. Pas de sprint. Le déplacement devient une action capitale."),
|
||||
(17, "buste", "Blessure interne : subit 2 CD de dégâts ignorant la RD à chaque fin de tour."),
|
||||
(20, "tete", "Perd ses actions normales au prochain tour. Difficulté +2 sur les tests basés sur la vue."),
|
||||
]
|
||||
|
||||
def _roll_crit_location(rng: random.Random) -> dict:
|
||||
roll = rng.randint(1, 20)
|
||||
for threshold, location, effect in _CRIT_LOCATION_TABLE:
|
||||
if roll <= threshold:
|
||||
return {"roll": roll, "location": location, "effect": effect}
|
||||
return {"roll": roll, "location": "buste", "effect": "Blessure grave."}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Modificateurs de portée (Résumé Écran MJ — table portées)
|
||||
# diff_modifier selon portée optimale de l'arme et distance réelle
|
||||
# ---------------------------------------------------------------------------
|
||||
_RANGE_DIFF: dict[tuple[str, str], int] = {
|
||||
# (portée_arme, distance_combat): modificateur difficulté
|
||||
("short", "short"): 0,
|
||||
("short", "medium"): 1,
|
||||
("short", "long"): 2,
|
||||
("short", "extreme"): 3,
|
||||
("medium", "short"): 1,
|
||||
("medium", "medium"): 0,
|
||||
("medium", "long"): 1,
|
||||
("medium", "extreme"): 2,
|
||||
("long", "short"): 2,
|
||||
("long", "medium"): 1,
|
||||
("long", "long"): 0,
|
||||
("long", "extreme"): 1,
|
||||
("extreme", "short"): 3,
|
||||
("extreme", "medium"): 2,
|
||||
("extreme", "long"): 1,
|
||||
("extreme", "extreme"): 0,
|
||||
}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Rang de compétence par défaut (utilisé si skills non trackés en DB)
|
||||
# ---------------------------------------------------------------------------
|
||||
_DEFAULT_SKILL_RANK: dict[str, int] = {
|
||||
"corps_a_corps": 2,
|
||||
"arme_legere": 2,
|
||||
"armes_lourdes": 2,
|
||||
"arme_energie": 1,
|
||||
}
|
||||
_ROLE_SKILL_BONUS: dict[str, dict[str, int]] = {
|
||||
"pillard": {"corps_a_corps": 1, "arme_legere": 1},
|
||||
"raider_boss":{"corps_a_corps": 2},
|
||||
"garde": {"arme_legere": 1, "corps_a_corps": 1},
|
||||
"eclaireur": {"arme_legere": 2},
|
||||
"technicien": {"arme_energie": 1},
|
||||
}
|
||||
|
||||
def _skill_rank(char: dict, skill: str) -> int:
|
||||
role = char.get("role", "survivant_generique")
|
||||
return (_DEFAULT_SKILL_RANK.get(skill, 1)
|
||||
+ _ROLE_SKILL_BONUS.get(role, {}).get(skill, 0))
|
||||
|
||||
def _melee_bonus_cd(strength: int) -> int:
|
||||
if strength < 7: return 0
|
||||
if strength <= 8: return 1
|
||||
if strength <= 10:return 2
|
||||
return 3
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Jets de base
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def roll_2d20(attribute: int, skill_rank: int, n_dice: int = 2,
|
||||
diff_modifier: int = 0) -> dict:
|
||||
"""
|
||||
Lance n_dice d20.
|
||||
Réussite : dé ≤ attribute + skill_rank
|
||||
Dé 1 = réussite critique (2 réussites). Dé 20 = complication.
|
||||
diff_modifier : s'ajoute à la difficulté cible (portée, couverture partielle…)
|
||||
"""
|
||||
sr = attribute + skill_rank
|
||||
rolls = [random.randint(1, 20) for _ in range(n_dice)]
|
||||
successes = sum(2 if r == 1 else 1 for r in rolls if r <= sr)
|
||||
complications = sum(1 for r in rolls if r == 20)
|
||||
return {
|
||||
"rolls": rolls, "sr": sr,
|
||||
"successes": successes, "complications": complications,
|
||||
"diff_modifier": diff_modifier,
|
||||
}
|
||||
|
||||
|
||||
def roll_cd(n_dice: int, rng: random.Random | None = None) -> dict:
|
||||
"""
|
||||
Lance n_dice dés de combat (d6).
|
||||
1→1dmg | 2→2dmg | 3-4→0 | 5-6→1dmg+1effet
|
||||
"""
|
||||
_r = rng or random
|
||||
damage = effects = 0
|
||||
rolls = []
|
||||
for _ in range(max(1, n_dice)):
|
||||
r = _r.randint(1, 6)
|
||||
rolls.append(r)
|
||||
if r == 1: damage += 1
|
||||
elif r == 2: damage += 2
|
||||
elif r >= 5: damage += 1; effects += 1
|
||||
return {"rolls": rolls, "damage": damage, "effects": effects}
|
||||
|
||||
|
||||
def _cover_reduction(cover_rating: int, rng: random.Random) -> int:
|
||||
"""
|
||||
Lance cover_rating CD, retourne les dégâts absorbés.
|
||||
(Résumé Écran MJ : Feuillage→1, Gravats/grillage→2, Béton/acier→3)
|
||||
"""
|
||||
if cover_rating <= 0:
|
||||
return 0
|
||||
cd = roll_cd(cover_rating, rng)
|
||||
return cd["damage"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Résolution des effets d'armes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _resolve_weapon_effects(effects_count: int, weapon_props: dict,
|
||||
base_damage: int, rng: random.Random) -> dict:
|
||||
"""
|
||||
Applique les effets de l'arme selon le nombre d'Effets obtenus.
|
||||
Retourne un dict avec les bonus/malus à appliquer.
|
||||
"""
|
||||
extra_damage = 0
|
||||
status_effects = []
|
||||
ammo_extra = 0
|
||||
|
||||
for _ in range(effects_count):
|
||||
if weapon_props.get("Brutal"):
|
||||
extra_damage += 1
|
||||
|
||||
if weapon_props.get("Etourdissant"):
|
||||
status_effects.append("etourdi")
|
||||
|
||||
if weapon_props.get("Persistant"):
|
||||
status_effects.append("persistant")
|
||||
|
||||
if weapon_props.get("Radioactif"):
|
||||
status_effects.append("radioactif_1rad")
|
||||
|
||||
if weapon_props.get("En_rafale"):
|
||||
status_effects.append("en_rafale_cible_proche")
|
||||
ammo_extra += 1
|
||||
|
||||
perforant = weapon_props.get("Perforant", 0)
|
||||
|
||||
return {
|
||||
"extra_damage": extra_damage,
|
||||
"perforant": perforant, # ignore X pts de RD par Effet
|
||||
"status_effects": status_effects,
|
||||
"ammo_extra": ammo_extra,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Point d'entrée principal
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def full_combat_round(
|
||||
attacker: dict,
|
||||
defender: dict,
|
||||
weapon: dict | None,
|
||||
combat_range: str = "medium",
|
||||
cover_rating: int = 0,
|
||||
rng: random.Random | None = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Résout un round de combat complet selon les règles Fallout 2D20.
|
||||
|
||||
Args:
|
||||
attacker : dict personnage attaquant (doit avoir les attributs SPECIAL)
|
||||
defender : dict personnage défenseur
|
||||
weapon : dict item équipé (None = mains nues)
|
||||
combat_range : distance réelle du combat ('short'|'medium'|'long'|'extreme')
|
||||
cover_rating : RD de la couverture (0-3)
|
||||
rng : Random optionnel pour reproductibilité
|
||||
|
||||
Returns:
|
||||
dict complet avec test, dégâts, effets, blessure critique éventuelle,
|
||||
munitions consommées, résumé narrative.
|
||||
"""
|
||||
_rng = rng or random.Random()
|
||||
|
||||
# --- Attribut + compétence selon type d'arme ---
|
||||
weapon_type = weapon.get("item_type", "weapon_melee") if weapon else "weapon_melee"
|
||||
weapon_range = weapon.get("damage_range", "short") if weapon else "short"
|
||||
weapon_props = weapon.get("properties") or {}
|
||||
if isinstance(weapon_props, str):
|
||||
import json
|
||||
weapon_props = json.loads(weapon_props)
|
||||
|
||||
if weapon_type == "weapon_melee" or weapon is None:
|
||||
attr_val = attacker["strength"]
|
||||
skill = "corps_a_corps"
|
||||
is_melee = True
|
||||
elif weapon_type == "weapon_energy":
|
||||
attr_val = attacker["perception"]
|
||||
skill = "arme_energie"
|
||||
is_melee = False
|
||||
else:
|
||||
if weapon_props.get("Lourd"):
|
||||
attr_val = attacker["endurance"]
|
||||
skill = "armes_lourdes"
|
||||
else:
|
||||
attr_val = attacker["agility"]
|
||||
skill = "arme_legere"
|
||||
is_melee = False
|
||||
|
||||
rank = _skill_rank(attacker, skill)
|
||||
|
||||
# --- Modificateur de portée ---
|
||||
range_mod = _RANGE_DIFF.get((weapon_range, combat_range), 0) if not is_melee else 0
|
||||
|
||||
# --- Défense cible ---
|
||||
difficulty = (2 if defender.get("agility", 5) >= 9 else 1) + range_mod
|
||||
|
||||
# --- Jet d'attaque ---
|
||||
test = roll_2d20(attr_val, rank, diff_modifier=range_mod)
|
||||
hit = test["successes"] >= difficulty
|
||||
|
||||
result: dict = {
|
||||
"hit": hit,
|
||||
"test": test,
|
||||
"difficulty": difficulty,
|
||||
"skill": skill,
|
||||
"weapon": weapon["item_slug"] if weapon else "mains_nues",
|
||||
"weapon_name": weapon["item_name"] if weapon else "Mains nues",
|
||||
"combat_range": combat_range,
|
||||
"cover_rating": cover_rating,
|
||||
"damage_raw": 0,
|
||||
"damage_cover": 0,
|
||||
"damage_final": 0,
|
||||
"extra_damage": 0,
|
||||
"effects": 0,
|
||||
"weapon_effects_applied": [],
|
||||
"critical_wound": None,
|
||||
"status_effects": [],
|
||||
"ammo_consumed": 0,
|
||||
"killed": False,
|
||||
"narrative": "",
|
||||
}
|
||||
|
||||
if not hit:
|
||||
result["narrative"] = (
|
||||
f"{attacker['name']} attaque {defender['name']} "
|
||||
f"({result['weapon_name']}) — RATÉ "
|
||||
f"[SR={test['sr']}, dés={test['rolls']}, diff={difficulty}]"
|
||||
)
|
||||
return result
|
||||
|
||||
# --- Dégâts bruts ---
|
||||
n_cd = (weapon.get("damage_dice", 1) if weapon else 1)
|
||||
if is_melee:
|
||||
n_cd += _melee_bonus_cd(attacker["strength"])
|
||||
|
||||
cd_result = roll_cd(n_cd, _rng)
|
||||
raw_damage = cd_result["damage"]
|
||||
effects_count = cd_result["effects"]
|
||||
|
||||
# --- Effets d'arme ---
|
||||
eff = _resolve_weapon_effects(effects_count, weapon_props, raw_damage, _rng)
|
||||
raw_damage += eff["extra_damage"]
|
||||
result["extra_damage"] = eff["extra_damage"]
|
||||
result["effects"] = effects_count
|
||||
result["weapon_effects_applied"] = eff["status_effects"]
|
||||
result["ammo_consumed"] = (0 if is_melee else 1) + eff["ammo_extra"]
|
||||
|
||||
# --- Couverture (RD) ---
|
||||
cover_absorbed = _cover_reduction(cover_rating, _rng)
|
||||
# Perforant réduit l'efficacité de la couverture
|
||||
perforant_total = eff["perforant"] * effects_count
|
||||
cover_absorbed = max(0, cover_absorbed - perforant_total)
|
||||
|
||||
final_damage = max(0, raw_damage - cover_absorbed)
|
||||
|
||||
result["damage_raw"] = raw_damage
|
||||
result["damage_cover"] = cover_absorbed
|
||||
result["damage_final"] = final_damage
|
||||
|
||||
# --- Blessure critique (≥5 dégâts après couverture) ---
|
||||
if final_damage >= 5:
|
||||
crit = _roll_crit_location(_rng)
|
||||
result["critical_wound"] = crit
|
||||
|
||||
# --- Application en DB ---
|
||||
new_hp = max(0, defender["hp"] - final_damage)
|
||||
db.update_character(defender["id"], hp=new_hp)
|
||||
|
||||
if new_hp == 0:
|
||||
db.update_character(defender["id"], is_alive=False)
|
||||
result["killed"] = True
|
||||
|
||||
# --- Radiations (effets Radioactif) ---
|
||||
rad_hits = result["weapon_effects_applied"].count("radioactif_1rad")
|
||||
if rad_hits > 0 and not defender.get("immune_radiation", False):
|
||||
new_rads = defender.get("rads", 0) + rad_hits
|
||||
db.update_character(defender["id"], rads=new_rads)
|
||||
result["rads_inflicted"] = rad_hits
|
||||
|
||||
# --- Statuts persistants (stockés dans pnj_state.last_actions) ---
|
||||
result["status_effects"] = [
|
||||
s for s in eff["status_effects"]
|
||||
if s not in ("radioactif_1rad", "en_rafale_cible_proche")
|
||||
]
|
||||
|
||||
# --- Narrative ---
|
||||
crit_txt = (
|
||||
f" [BLESSURE CRITIQUE — {result['critical_wound']['location'].upper()} : "
|
||||
f"{result['critical_wound']['effect']}]"
|
||||
if result["critical_wound"] else ""
|
||||
)
|
||||
killed_txt = " [MORT]" if result["killed"] else f" → {new_hp}/{defender['max_hp']} PV"
|
||||
cover_txt = f" (couverture absorbe {cover_absorbed})" if cover_absorbed else ""
|
||||
eff_txt = f" +{effects_count} effets" if effects_count else ""
|
||||
|
||||
result["narrative"] = (
|
||||
f"{attacker['name']} touche {defender['name']} "
|
||||
f"({result['weapon_name']}) : "
|
||||
f"{raw_damage} brut{cover_txt} = {final_damage} dégâts{eff_txt}"
|
||||
f"{killed_txt}{crit_txt}"
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Intégration avec actions.py (remplace resolve_attack)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def resolve_attack(attacker: dict, defender: dict,
|
||||
weapon: dict | None,
|
||||
session_id: int, day: int, tick: int,
|
||||
combat_range: str = "medium",
|
||||
cover_rating: int = 0,
|
||||
rng: random.Random | None = None) -> dict:
|
||||
"""
|
||||
Wrapper appelable depuis actions.py et encounter.py.
|
||||
Lance full_combat_round et logue le résultat en DB.
|
||||
"""
|
||||
result = full_combat_round(attacker, defender, weapon,
|
||||
combat_range, cover_rating, rng)
|
||||
|
||||
db.log_event(
|
||||
session_id=session_id, day=day, tick=tick,
|
||||
event_type="combat",
|
||||
description=result["narrative"],
|
||||
actor_id=attacker["id"],
|
||||
location_slug=attacker.get("location_slug"),
|
||||
mechanical_effect={
|
||||
"hit": result["hit"],
|
||||
"damage_final": result["damage_final"],
|
||||
"damage_raw": result["damage_raw"],
|
||||
"damage_cover": result["damage_cover"],
|
||||
"effects": result["effects"],
|
||||
"weapon": result["weapon"],
|
||||
"critical_wound":result["critical_wound"],
|
||||
"status_effects":result["status_effects"],
|
||||
"ammo_consumed": result["ammo_consumed"],
|
||||
"killed": result["killed"],
|
||||
}
|
||||
)
|
||||
return result
|
||||
+15
-2
@@ -16,6 +16,7 @@ import os
|
||||
import random
|
||||
import db
|
||||
import actions as act_module
|
||||
import combat_engine
|
||||
|
||||
_DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data")
|
||||
|
||||
@@ -538,7 +539,19 @@ def resolve_encounter(
|
||||
outcomes_cfg = tables["outcomes"]
|
||||
|
||||
if reaction == "combat":
|
||||
result = _combat_roll(char, entity, rng)
|
||||
# Utilise le moteur complet si l'entité est un PNJ en DB (actif/boss)
|
||||
if entity.get("db_id") and entity.get("tier") in ("actif", "boss"):
|
||||
defender_full = db.get_character_by_id(entity["db_id"])
|
||||
inv_att = []
|
||||
with db.cursor() as cur:
|
||||
cur.execute("SELECT * FROM inventory WHERE character_id=%s AND session_id=%s", (char["id"], session_id))
|
||||
inv_att = [dict(r) for r in cur.fetchall()]
|
||||
weapon = act_module._get_equipped_weapon(inv_att)
|
||||
cr = combat_engine.full_combat_round(char, defender_full, weapon, rng=rng)
|
||||
result = "win" if cr["hit"] and not cr["killed"] else ("win" if cr["killed"] else "loss")
|
||||
outcome = result
|
||||
else:
|
||||
result = _combat_roll(char, entity, rng)
|
||||
outcome = result
|
||||
cfg = outcomes_cfg["combat"][result]
|
||||
|
||||
@@ -675,7 +688,7 @@ def process_encounters_tick(session_id: int, day: int, tick: int, rng_seed: int
|
||||
day = day,
|
||||
tick = tick,
|
||||
event_type = f"encounter_{tier}",
|
||||
character_id = char["id"],
|
||||
actor_id = char["id"],
|
||||
location_slug = zone,
|
||||
description = result["narrative"],
|
||||
mechanical_effect = {
|
||||
|
||||
@@ -0,0 +1,374 @@
|
||||
"""
|
||||
Génération RAG+LLM d'un batch de PNJ pour la simulation.
|
||||
Lance HORS simulation : python generate_pnj_batch.py
|
||||
|
||||
Pipeline :
|
||||
1. Query ChromaDB (RAG) pour règles création personnage Fallout 2D20
|
||||
2. Prompt qwen2.5:14b avec contexte + structure JSON attendue
|
||||
3. Parse + validation des PNJ générés
|
||||
4. Insertion en DB via character_creator + pnj_factory
|
||||
|
||||
Ce script est one-shot — il ne tourne pas pendant la simulation.
|
||||
"""
|
||||
|
||||
import sys, os, json, re, time, requests
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
import character_creator as cc
|
||||
import pnj_factory as factory
|
||||
import db
|
||||
from config import SESSION_ID
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
CHROMA_URL = "http://localhost:8000" # ChromaDB REST API
|
||||
OLLAMA_URL = "http://localhost:11434"
|
||||
LLM_MODEL = "qwen2.5:14b"
|
||||
COLLECTION = "fallout_vst"
|
||||
|
||||
# Combien de PNJ à générer par appel
|
||||
N_PNJ_TIER2 = 5 # PNJ+ nommés (lieutenants, personnages récurrents)
|
||||
N_PNJ_TIER3 = 10 # PNJ de passage (archétypes instanciés)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RAG — Récupération contexte depuis ChromaDB
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def rag_query(query: str, n_results: int = 5) -> str:
|
||||
"""Interroge ChromaDB et retourne le contexte textuel concatené."""
|
||||
try:
|
||||
resp = requests.post(
|
||||
f"{CHROMA_URL}/api/v2/collections/{COLLECTION}/query",
|
||||
json={"query_texts": [query], "n_results": n_results},
|
||||
timeout=15,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
docs = data.get("documents", [[]])[0]
|
||||
return "\n\n---\n\n".join(docs) if docs else ""
|
||||
except Exception as e:
|
||||
print(f" [RAG WARN] ChromaDB inaccessible : {e}")
|
||||
return ""
|
||||
|
||||
|
||||
def build_rag_context() -> str:
|
||||
"""Combine plusieurs queries RAG pour un contexte riche."""
|
||||
queries = [
|
||||
"création personnage Fallout 2D20 SPECIAL attributs statistiques",
|
||||
"origines survivant habitant abri initié confrerie goule super mutant",
|
||||
"rôles narratifs pillard marchand garde éclaireur docteur technicien",
|
||||
"factions Louisiana Nouvelle-Orléans post-apocalyptique",
|
||||
"équipements armes armures Fallout terres désolées bayou",
|
||||
]
|
||||
parts = []
|
||||
for q in queries:
|
||||
ctx = rag_query(q, n_results=3)
|
||||
if ctx:
|
||||
parts.append(ctx)
|
||||
time.sleep(0.2)
|
||||
return "\n\n=====\n\n".join(parts)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# LLM — Génération des PNJ
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
PROMPT_TIER2 = """Tu es le Maître de Jeu de "Fallout : Venice of Wasteland", un JdR 2D20 post-apocalyptique en Louisiane.
|
||||
|
||||
CONTEXTE DES RÈGLES (extrait du manuel) :
|
||||
{rag_context}
|
||||
|
||||
FACTIONS EXISTANTES : union, cda, syndicat_capitole, regie, dynaste_oak, ecumeurs, consortium, grand_krewe
|
||||
|
||||
ZONES : independance, laplace, baton_rouge, oak_plantation, la_paroisse, nola_vieux_carre, nola_cbd, nola_fleuve, pearl_river, donaldsonville
|
||||
|
||||
RÔLES DISPONIBLES : garde, marchand, eclaireur, docteur, technicien, pillard, lieutenant, chasseur_primes, passeur, scavenger, survivant_generique
|
||||
|
||||
ORIGINES DISPONIBLES : survivant, habitant_abri, initie_confrerie, goule, super_mutant
|
||||
|
||||
Génère exactement {n} PNJ de TIER 2 (personnages nommés secondaires, lieutenants et personnages récurrents importants).
|
||||
Chaque PNJ doit avoir :
|
||||
- Un nom louisianais immersif avec un surnom entre guillemets
|
||||
- Un rôle cohérent avec sa faction
|
||||
- Des stats SPECIAL réalistes (1-10, super_mutant peut avoir FOR/END jusqu'à 12)
|
||||
- Une courte description de personnalité (1 phrase)
|
||||
|
||||
Réponds UNIQUEMENT avec un JSON valide, aucun texte avant ou après :
|
||||
{{
|
||||
"pnj": [
|
||||
{{
|
||||
"name": "Prénom 'Surnom' Nom",
|
||||
"role": "garde",
|
||||
"origin": "survivant",
|
||||
"faction_slug": "union",
|
||||
"location_slug": "independance",
|
||||
"strength": 7,
|
||||
"perception": 5,
|
||||
"endurance": 6,
|
||||
"charisma": 4,
|
||||
"intelligence": 5,
|
||||
"agility": 6,
|
||||
"luck": 5,
|
||||
"caps": 150,
|
||||
"description": "Une phrase de description"
|
||||
}}
|
||||
]
|
||||
}}"""
|
||||
|
||||
PROMPT_TIER3 = """Tu es le Maître de Jeu de "Fallout : Venice of Wasteland".
|
||||
|
||||
ARCHÉTYPES DISPONIBLES (clés pour instanciation) :
|
||||
{archetypes}
|
||||
|
||||
ZONES : independance, laplace, baton_rouge, oak_plantation, la_paroisse, nola_vieux_carre, nola_cbd, nola_fleuve, pearl_river, donaldsonville
|
||||
|
||||
FACTIONS : union, cda, syndicat_capitole, regie, dynaste_oak, ecumeurs, consortium, grand_krewe (ou null pour sans faction)
|
||||
|
||||
Génère exactement {n} PNJ de TIER 3 (personnages de passage, vendeurs, voyageurs, rencontres).
|
||||
Chaque PNJ est instancié depuis un archétype. Choisis des archétypes variés et des localisations cohérentes.
|
||||
|
||||
Réponds UNIQUEMENT avec un JSON valide :
|
||||
{{
|
||||
"pnj": [
|
||||
{{
|
||||
"archetype": "vendeur_ambulant",
|
||||
"name": "Prénom 'Surnom' Nom",
|
||||
"faction_slug": "consortium",
|
||||
"location_slug": "laplace"
|
||||
}}
|
||||
]
|
||||
}}"""
|
||||
|
||||
|
||||
def call_llm(prompt: str, timeout: int = 120) -> str:
|
||||
"""Appelle Ollama et retourne le texte généré."""
|
||||
resp = requests.post(
|
||||
f"{OLLAMA_URL}/api/generate",
|
||||
json={"model": LLM_MODEL, "prompt": prompt, "stream": False},
|
||||
timeout=timeout,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.json().get("response", "")
|
||||
|
||||
|
||||
def parse_json_response(text: str) -> dict | None:
|
||||
"""Extrait le JSON d'une réponse LLM."""
|
||||
match = re.search(r"\{.*\}", text, re.DOTALL)
|
||||
if not match:
|
||||
return None
|
||||
try:
|
||||
return json.loads(match.group())
|
||||
except json.JSONDecodeError as e:
|
||||
print(f" [PARSE ERR] {e}")
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Validation et normalisation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_VALID_ROLES = set(cc._ROLES.keys())
|
||||
_VALID_ORIGINS = set(cc.ORIGINS.keys())
|
||||
|
||||
def validate_pnj_tier2(pnj: dict) -> dict | None:
|
||||
"""Valide et normalise un PNJ tier 2 généré par le LLM."""
|
||||
required = ["name", "role", "origin", "faction_slug", "location_slug",
|
||||
"strength", "perception", "endurance", "charisma",
|
||||
"intelligence", "agility", "luck"]
|
||||
for field in required:
|
||||
if field not in pnj:
|
||||
print(f" [SKIP] Champ manquant '{field}' pour {pnj.get('name','?')}")
|
||||
return None
|
||||
|
||||
if pnj["role"] not in _VALID_ROLES:
|
||||
pnj["role"] = "survivant_generique"
|
||||
if pnj["origin"] not in _VALID_ORIGINS:
|
||||
pnj["origin"] = "survivant"
|
||||
|
||||
# Clamper les stats 1-10 (12 pour super_mutant)
|
||||
cap = 12 if pnj["origin"] == "super_mutant" else 10
|
||||
for attr in ["strength","perception","endurance","charisma","intelligence","agility","luck"]:
|
||||
pnj[attr] = max(1, min(cap, int(pnj[attr])))
|
||||
|
||||
pnj.setdefault("caps", 200)
|
||||
return pnj
|
||||
|
||||
|
||||
def validate_pnj_tier3(pnj: dict, valid_archetypes: set) -> dict | None:
|
||||
"""Valide un spec PNJ tier 3."""
|
||||
if "archetype" not in pnj or pnj["archetype"] not in valid_archetypes:
|
||||
print(f" [SKIP] Archétype invalide '{pnj.get('archetype','?')}'")
|
||||
return None
|
||||
if "name" not in pnj:
|
||||
print(f" [SKIP] Pas de nom pour archétype {pnj['archetype']}")
|
||||
return None
|
||||
return pnj
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Insertion en DB
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def insert_tier2(pnj_data: dict, seed: int) -> dict | None:
|
||||
"""Insère un PNJ tier 2 généré par LLM en DB avec équipement."""
|
||||
try:
|
||||
# Calcul HP selon règles (END + LCK)
|
||||
max_hp = pnj_data["endurance"] + pnj_data["luck"]
|
||||
char_dict = {
|
||||
"session_id": SESSION_ID,
|
||||
"name": pnj_data["name"],
|
||||
"character_type": "pnj_sim",
|
||||
"faction_slug": pnj_data.get("faction_slug"),
|
||||
"location_slug": pnj_data.get("location_slug"),
|
||||
"seed_base": seed,
|
||||
"role": pnj_data["role"],
|
||||
"origin": pnj_data["origin"],
|
||||
"strength": pnj_data["strength"],
|
||||
"perception": pnj_data["perception"],
|
||||
"endurance": pnj_data["endurance"],
|
||||
"charisma": pnj_data["charisma"],
|
||||
"intelligence": pnj_data["intelligence"],
|
||||
"agility": pnj_data["agility"],
|
||||
"luck": pnj_data["luck"],
|
||||
"hp": max_hp,
|
||||
"max_hp": max_hp,
|
||||
"caps": pnj_data.get("caps", 200),
|
||||
"rads": 0,
|
||||
"is_alive": True,
|
||||
"is_in_shelter": False,
|
||||
"needs_food": cc.ORIGINS[pnj_data["origin"]]["needs_food"],
|
||||
"needs_water": cc.ORIGINS[pnj_data["origin"]]["needs_water"],
|
||||
"needs_sleep": cc.ORIGINS[pnj_data["origin"]]["needs_sleep"],
|
||||
"immune_radiation": cc.ORIGINS[pnj_data["origin"]]["immune_radiation"],
|
||||
"immune_poison": cc.ORIGINS[pnj_data["origin"]]["immune_poison"],
|
||||
"description": pnj_data.get("description", ""),
|
||||
"tier": 2,
|
||||
"archetype_key": None,
|
||||
}
|
||||
char_id = cc._insert_character(char_dict)
|
||||
char_dict["id"] = char_id
|
||||
items = factory.equip_character(char_id, pnj_data["role"], SESSION_ID, seed=seed)
|
||||
return {"char": char_dict, "items": items}
|
||||
except Exception as e:
|
||||
print(f" [DB ERR] {pnj_data.get('name','?')} : {e}")
|
||||
return None
|
||||
|
||||
|
||||
def insert_tier3(spec: dict, seed: int) -> dict | None:
|
||||
"""Instancie un archétype tier 3 en DB."""
|
||||
try:
|
||||
char = factory.spawn_from_archetype(
|
||||
archetype_key = spec["archetype"],
|
||||
session_id = SESSION_ID,
|
||||
name = spec["name"],
|
||||
faction_slug = spec.get("faction_slug"),
|
||||
location_slug = spec.get("location_slug"),
|
||||
seed = seed,
|
||||
tier_override = 3,
|
||||
)
|
||||
return char
|
||||
except Exception as e:
|
||||
print(f" [DB ERR] {spec.get('name','?')} : {e}")
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def main():
|
||||
print("=== Génération RAG+LLM de PNJ ===\n")
|
||||
|
||||
# Récupération contexte RAG
|
||||
print("1. Query RAG (ChromaDB)...")
|
||||
rag_ctx = build_rag_context()
|
||||
if rag_ctx:
|
||||
print(f" Contexte récupéré : {len(rag_ctx)} caractères\n")
|
||||
else:
|
||||
print(" Aucun contexte RAG — le LLM utilisera ses connaissances internes\n")
|
||||
|
||||
# Archetypes disponibles pour tier 3
|
||||
archetypes_list = factory.list_archetypes()
|
||||
archetypes_str = "\n".join(
|
||||
f" - {a['key']} : {a['description']}" for a in archetypes_list
|
||||
)
|
||||
valid_archetype_keys = {a["key"] for a in archetypes_list}
|
||||
|
||||
# ---- Tier 2 ----
|
||||
print(f"2. Génération {N_PNJ_TIER2} PNJ+ tier 2 (qwen2.5:14b)...")
|
||||
prompt2 = PROMPT_TIER2.format(
|
||||
rag_context=rag_ctx[:3000] if rag_ctx else "(manuel non disponible, utilise tes connaissances)",
|
||||
n=N_PNJ_TIER2,
|
||||
)
|
||||
try:
|
||||
t0 = time.time()
|
||||
resp2 = call_llm(prompt2, timeout=180)
|
||||
print(f" LLM répondu en {time.time()-t0:.1f}s")
|
||||
data2 = parse_json_response(resp2)
|
||||
except Exception as e:
|
||||
print(f" [LLM ERR] {e}")
|
||||
data2 = None
|
||||
|
||||
tier2_created = []
|
||||
if data2 and "pnj" in data2:
|
||||
for i, pnj in enumerate(data2["pnj"]):
|
||||
validated = validate_pnj_tier2(pnj)
|
||||
if not validated:
|
||||
continue
|
||||
result = insert_tier2(validated, seed=3000 + i)
|
||||
if result:
|
||||
char = result["char"]
|
||||
tier2_created.append(char)
|
||||
print(f" ✓ [T2] {char['name']:<40} {char.get('faction_slug','—'):<20} @ {char.get('location_slug','—')}")
|
||||
print(f" Items: {', '.join(result['items'])}")
|
||||
else:
|
||||
print(" [WARN] Pas de PNJ tier 2 générés")
|
||||
|
||||
# ---- Tier 3 ----
|
||||
print(f"\n3. Génération {N_PNJ_TIER3} PNJ de passage tier 3 (qwen2.5:14b)...")
|
||||
prompt3 = PROMPT_TIER3.format(
|
||||
archetypes=archetypes_str,
|
||||
n=N_PNJ_TIER3,
|
||||
)
|
||||
try:
|
||||
t0 = time.time()
|
||||
resp3 = call_llm(prompt3, timeout=180)
|
||||
print(f" LLM répondu en {time.time()-t0:.1f}s")
|
||||
data3 = parse_json_response(resp3)
|
||||
except Exception as e:
|
||||
print(f" [LLM ERR] {e}")
|
||||
data3 = None
|
||||
|
||||
tier3_created = []
|
||||
if data3 and "pnj" in data3:
|
||||
for i, spec in enumerate(data3["pnj"]):
|
||||
validated = validate_pnj_tier3(spec, valid_archetype_keys)
|
||||
if not validated:
|
||||
continue
|
||||
char = insert_tier3(validated, seed=4000 + i)
|
||||
if char:
|
||||
tier3_created.append(char)
|
||||
print(f" ✓ [T3] {char.get('name','?'):<40} archetype={spec['archetype']:<20} @ {spec.get('location_slug','—')}")
|
||||
else:
|
||||
print(" [WARN] Pas de PNJ tier 3 générés")
|
||||
|
||||
# ---- Résumé ----
|
||||
print(f"\n=== Résumé ===")
|
||||
print(f" Tier 2 créés : {len(tier2_created)}")
|
||||
print(f" Tier 3 créés : {len(tier3_created)}")
|
||||
|
||||
all_chars = db.get_all_characters(SESSION_ID)
|
||||
by_tier = {}
|
||||
for c in all_chars:
|
||||
t = c.get("tier") or 1
|
||||
by_tier[t] = by_tier.get(t, 0) + 1
|
||||
labels = {0:"BOSS", 1:"ACTIFS SIM", 2:"PNJ+", 3:"PASSAGE"}
|
||||
print("\n DB total par tier :")
|
||||
for t in sorted(by_tier):
|
||||
print(f" Tier {t} ({labels.get(t,'?')}) : {by_tier[t]} PNJ")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user