Spaces:
Running on Zero
Running on Zero
| """ | |
| Condition Evaluator - Logic system for game config conditions and transitions. | |
| Provides: | |
| - ConditionEvaluator: Evaluate condition expressions against GameState | |
| - TransitionResolver: Resolve dynamic transitions (random, conditional) | |
| - EffectApplicator: Apply declarative effects to GameState | |
| """ | |
| from typing import Any, Dict, List, Optional, Union | |
| import random | |
| from game_state import GameState | |
| class ConditionEvaluator: | |
| """ | |
| Evaluates condition expressions against GameState. | |
| Supports: | |
| - Atomic conditions: has_item, met_person, flag, visited, mission_*, money, counter, knowledge | |
| - Compound conditions: and, or, not | |
| - Numeric comparisons: gte, lte, gt, lt, eq | |
| """ | |
| def __init__(self, game_state: GameState): | |
| self.state = game_state | |
| def evaluate(self, condition: Any) -> bool: | |
| """ | |
| Evaluate a condition expression. | |
| Args: | |
| condition: Can be: | |
| - None/empty dict: Always True (no condition) | |
| - str: Flag name to check | |
| - dict: Condition expression | |
| Returns: | |
| bool: Whether condition is satisfied | |
| """ | |
| # No condition = always true (backwards compatible) | |
| if condition is None or condition == {}: | |
| return True | |
| # Simple string = flag check | |
| if isinstance(condition, str): | |
| return self.state.has_flag(condition) | |
| if not isinstance(condition, dict): | |
| return False | |
| # Compound operators | |
| if "and" in condition: | |
| return all(self.evaluate(c) for c in condition["and"]) | |
| if "or" in condition: | |
| return any(self.evaluate(c) for c in condition["or"]) | |
| if "not" in condition: | |
| return not self.evaluate(condition["not"]) | |
| # Atomic conditions | |
| return self._evaluate_atomic(condition) | |
| def _evaluate_atomic(self, condition: Dict) -> bool: | |
| """Evaluate a single atomic condition.""" | |
| # ==================== Item Checks ==================== | |
| if "has_item" in condition: | |
| return self.state.has_item(condition["has_item"]) | |
| if "not_has_item" in condition: | |
| return not self.state.has_item(condition["not_has_item"]) | |
| # ==================== Person Checks ==================== | |
| if "met_person" in condition: | |
| return self.state.has_met(condition["met_person"]) | |
| if "not_met_person" in condition: | |
| return not self.state.has_met(condition["not_met_person"]) | |
| # ==================== Flag Checks ==================== | |
| if "flag" in condition: | |
| return self.state.has_flag(condition["flag"]) | |
| if "not_flag" in condition: | |
| return not self.state.has_flag(condition["not_flag"]) | |
| # ==================== Location Checks ==================== | |
| if "visited" in condition: | |
| return self.state.has_visited(condition["visited"]) | |
| if "not_visited" in condition: | |
| return not self.state.has_visited(condition["not_visited"]) | |
| if "discovered" in condition: | |
| return self.state.has_discovered(condition["discovered"]) | |
| # ==================== Mission Checks ==================== | |
| if "mission_complete" in condition: | |
| return self.state.is_mission_complete(condition["mission_complete"]) | |
| if "mission_active" in condition: | |
| return self.state.is_mission_active(condition["mission_active"]) | |
| if "mission_failed" in condition: | |
| return self.state.is_mission_failed(condition["mission_failed"]) | |
| # ==================== Money Comparison ==================== | |
| if "money" in condition: | |
| return self._compare_numeric(self.state.money, condition["money"]) | |
| # ==================== Counter Comparison ==================== | |
| if "counter" in condition: | |
| counter_cond = condition["counter"] | |
| for counter_name, comparison in counter_cond.items(): | |
| value = self.state.get_counter(counter_name) | |
| if not self._compare_numeric(value, comparison): | |
| return False | |
| return True | |
| # ==================== Knowledge Checks ==================== | |
| if "knowledge" in condition: | |
| return self.state.has_knowledge(condition["knowledge"]) | |
| if "knowledge_value" in condition: | |
| kv = condition["knowledge_value"] | |
| key = kv.get("key") | |
| actual = self.state.get_knowledge(key) | |
| if "eq" in kv: | |
| return actual == kv["eq"] | |
| if "neq" in kv: | |
| return actual != kv["neq"] | |
| return False | |
| # ==================== Reputation Check ==================== | |
| if "reputation" in condition: | |
| rep_cond = condition["reputation"] | |
| npc = rep_cond.get("npc") | |
| actual = self.state.get_reputation(npc) | |
| return self._compare_numeric(actual, rep_cond) | |
| # ==================== Visit Count Check ==================== | |
| if "visit_count" in condition: | |
| vc = condition["visit_count"] | |
| state_key = vc.get("state") | |
| actual = self.state.get_visit_count(state_key) | |
| return self._compare_numeric(actual, vc) | |
| # Unknown condition type - return False (safe default) | |
| return False | |
| def _compare_numeric(self, actual: int, comparison: Any) -> bool: | |
| """ | |
| Evaluate numeric comparisons. | |
| comparison can be: | |
| - int: exact match | |
| - {"gte": n}: >= | |
| - {"lte": n}: <= | |
| - {"gt": n}: > | |
| - {"lt": n}: < | |
| - {"eq": n}: == | |
| - {"neq": n}: != | |
| """ | |
| if isinstance(comparison, (int, float)): | |
| return actual == comparison | |
| if isinstance(comparison, dict): | |
| if "gte" in comparison: | |
| return actual >= comparison["gte"] | |
| if "lte" in comparison: | |
| return actual <= comparison["lte"] | |
| if "gt" in comparison: | |
| return actual > comparison["gt"] | |
| if "lt" in comparison: | |
| return actual < comparison["lt"] | |
| if "eq" in comparison: | |
| return actual == comparison["eq"] | |
| if "neq" in comparison: | |
| return actual != comparison["neq"] | |
| return False | |
| class TransitionResolver: | |
| """ | |
| Resolves transition specifications to concrete target states. | |
| Handles deterministic, random, and conditional transitions. | |
| """ | |
| def __init__(self, game_state: GameState): | |
| self.state = game_state | |
| self.evaluator = ConditionEvaluator(game_state) | |
| def resolve(self, transition: Any) -> str: | |
| """ | |
| Resolve a transition specification to a target state. | |
| Args: | |
| transition: Can be: | |
| - str: Direct target (current behavior, deterministic) | |
| - dict: Complex transition spec (random, conditional) | |
| Returns: | |
| str: Target state name | |
| Raises: | |
| ValueError: If transition format is invalid or no condition matches | |
| """ | |
| # Simple string = deterministic transition (backwards compatible) | |
| if isinstance(transition, str): | |
| return transition | |
| if not isinstance(transition, dict): | |
| raise ValueError(f"Invalid transition type: {type(transition)}") | |
| # Weighted random: {"random": [["state_a", 0.7], ["state_b", 0.3]]} | |
| if "random" in transition: | |
| return self._resolve_weighted_random(transition["random"]) | |
| # Equal-weight pool: {"random_from": ["a", "b", "c"]} | |
| if "random_from" in transition: | |
| pool = transition["random_from"] | |
| if not pool: | |
| raise ValueError("random_from pool is empty") | |
| return random.choice(pool) | |
| # Simple conditional: {"if": condition, "then": target, "else": fallback} | |
| if "if" in transition: | |
| condition = transition["if"] | |
| if self.evaluator.evaluate(condition): | |
| then_target = transition.get("then") | |
| if then_target: | |
| return self.resolve(then_target) | |
| else: | |
| else_target = transition.get("else") | |
| if else_target: | |
| return self.resolve(else_target) | |
| # If no matching branch, this is an error | |
| raise ValueError("Conditional transition has no matching branch") | |
| # Chained conditions: {"conditions": [{if, then}, {if, then}, {default}]} | |
| if "conditions" in transition: | |
| for cond_block in transition["conditions"]: | |
| # Default case (no condition) | |
| if "default" in cond_block: | |
| return self.resolve(cond_block["default"]) | |
| # Conditional case | |
| if "if" in cond_block and self.evaluator.evaluate(cond_block["if"]): | |
| return self.resolve(cond_block["then"]) | |
| # No condition matched and no default | |
| raise ValueError("No condition matched and no default provided") | |
| raise ValueError(f"Unknown transition format: {transition}") | |
| def _resolve_weighted_random(self, weights: List) -> str: | |
| """ | |
| Select from weighted random options. | |
| Args: | |
| weights: List of [state, probability] pairs | |
| Returns: | |
| Selected state name | |
| """ | |
| if not weights: | |
| raise ValueError("Weighted random list is empty") | |
| states = [w[0] for w in weights] | |
| probs = [w[1] for w in weights] | |
| # Normalize probabilities if they don't sum to 1 | |
| total = sum(probs) | |
| if total <= 0: | |
| raise ValueError("Weights must sum to positive number") | |
| if abs(total - 1.0) > 0.001: | |
| probs = [p / total for p in probs] | |
| return random.choices(states, weights=probs, k=1)[0] | |
| class EffectApplicator: | |
| """ | |
| Applies declarative effect specifications to GameState. | |
| Supports: | |
| - Items: add_item, remove_item | |
| - Money: add_money, remove_money | |
| - People: add_person | |
| - Locations: add_location | |
| - Flags: set_flag, clear_flag | |
| - Counters: set_counter, increment, decrement | |
| - Knowledge: set_knowledge | |
| - Missions: start_mission, complete_mission, fail_mission | |
| - Reputation: adjust_reputation | |
| """ | |
| def __init__(self, game_state: GameState): | |
| self.state = game_state | |
| def apply(self, effects: Dict) -> None: | |
| """ | |
| Apply a set of effects to the game state. | |
| Args: | |
| effects: Dict of effect specifications | |
| """ | |
| if not effects: | |
| return | |
| # ==================== Item Effects ==================== | |
| if "add_item" in effects: | |
| item = effects["add_item"] | |
| if isinstance(item, list): | |
| self.state.add_items(item) | |
| else: | |
| self.state.add_item(item) | |
| if "remove_item" in effects: | |
| item = effects["remove_item"] | |
| if isinstance(item, list): | |
| for i in item: | |
| self.state.remove_item(i) | |
| else: | |
| self.state.remove_item(item) | |
| # ==================== Money Effects ==================== | |
| if "add_money" in effects: | |
| self.state.add_money(effects["add_money"]) | |
| if "remove_money" in effects: | |
| self.state.remove_money(effects["remove_money"]) | |
| if "set_money" in effects: | |
| self.state.money = effects["set_money"] | |
| # ==================== Person Effects ==================== | |
| if "add_person" in effects: | |
| person = effects["add_person"] | |
| if isinstance(person, list): | |
| for p in person: | |
| self.state.meet_person(p) | |
| else: | |
| self.state.meet_person(person) | |
| # ==================== Location Effects ==================== | |
| if "add_location" in effects: | |
| location = effects["add_location"] | |
| if isinstance(location, list): | |
| for loc in location: | |
| self.state.discover_location(loc) | |
| else: | |
| self.state.discover_location(location) | |
| if "visit_location" in effects: | |
| location = effects["visit_location"] | |
| if isinstance(location, list): | |
| for loc in location: | |
| self.state.visit_location(loc) | |
| else: | |
| self.state.visit_location(location) | |
| # ==================== Flag Effects ==================== | |
| if "set_flag" in effects: | |
| flag = effects["set_flag"] | |
| if isinstance(flag, list): | |
| for f in flag: | |
| self.state.set_flag(f, True) | |
| elif isinstance(flag, dict): | |
| for f, v in flag.items(): | |
| self.state.set_flag(f, v) | |
| else: | |
| self.state.set_flag(flag, True) | |
| if "clear_flag" in effects: | |
| flag = effects["clear_flag"] | |
| if isinstance(flag, list): | |
| for f in flag: | |
| self.state.clear_flag(f) | |
| else: | |
| self.state.clear_flag(flag) | |
| if "toggle_flag" in effects: | |
| flag = effects["toggle_flag"] | |
| if isinstance(flag, list): | |
| for f in flag: | |
| self.state.toggle_flag(f) | |
| else: | |
| self.state.toggle_flag(flag) | |
| # ==================== Counter Effects ==================== | |
| if "set_counter" in effects: | |
| for name, value in effects["set_counter"].items(): | |
| self.state.set_counter(name, value) | |
| if "increment" in effects: | |
| for name, amount in effects["increment"].items(): | |
| self.state.increment_counter(name, amount) | |
| if "decrement" in effects: | |
| for name, amount in effects["decrement"].items(): | |
| self.state.decrement_counter(name, amount) | |
| # ==================== Knowledge Effects ==================== | |
| if "set_knowledge" in effects: | |
| for key, value in effects["set_knowledge"].items(): | |
| self.state.update_knowledge(key, value) | |
| if "remove_knowledge" in effects: | |
| key = effects["remove_knowledge"] | |
| if isinstance(key, list): | |
| for k in key: | |
| self.state.remove_knowledge(k) | |
| else: | |
| self.state.remove_knowledge(key) | |
| # ==================== Mission Effects ==================== | |
| if "start_mission" in effects: | |
| mission = effects["start_mission"] | |
| if isinstance(mission, str): | |
| self.state.start_mission(mission) | |
| elif isinstance(mission, dict): | |
| for m_id, m_data in mission.items(): | |
| self.state.start_mission(m_id, m_data if isinstance(m_data, dict) else None) | |
| elif isinstance(mission, list): | |
| for m in mission: | |
| self.state.start_mission(m) | |
| if "complete_mission" in effects: | |
| mission = effects["complete_mission"] | |
| if isinstance(mission, list): | |
| for m in mission: | |
| self.state.complete_mission(m) | |
| else: | |
| self.state.complete_mission(mission) | |
| if "fail_mission" in effects: | |
| mission = effects["fail_mission"] | |
| if isinstance(mission, list): | |
| for m in mission: | |
| self.state.fail_mission(m) | |
| else: | |
| self.state.fail_mission(mission) | |
| if "update_mission" in effects: | |
| for mission_id, updates in effects["update_mission"].items(): | |
| self.state.update_mission(mission_id, updates) | |
| # ==================== Reputation Effects ==================== | |
| if "adjust_reputation" in effects: | |
| for npc, change in effects["adjust_reputation"].items(): | |
| self.state.adjust_reputation(npc, change) | |
| if "set_reputation" in effects: | |
| for npc, value in effects["set_reputation"].items(): | |
| self.state.npc_reputation[npc] = value | |