nopcoder/game_runtime.py at main · dotpipe/nopcoder · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
from game_player import GamePlayer
from game_trainer import GameTrainer

from game_metrics import GameMetrics
from game_balancer import GameBalancer
from balance_applier import BalanceApplier

from game_learning_layers import GameLearningLayers
from structured_game_mutator import StructuredGameMutator
class GameRuntime:

    def __init__(self, controller, game_dsl, compiler, balancer, metrics, applier):

        self.controller = controller
        self.game_dsl = game_dsl
        self.compiler = compiler

        self.balancer = balancer
        self.metrics = metrics
        self.applier = applier

        self.game_player = GamePlayer(game_dsl)
        self.game_trainer = GameTrainer(controller.model_router)

    def run(self, goal, input_text, game_name):

        game = self.game_dsl.get(game_name)

        if not game:
            return {"error": "game not found"}

        rules = self.compiler.compile(
            game,
            self.controller.world,
            self.controller.plugin_runtime
        )

        step = 0

        events, rewards = [], {}

        while step < 500:

            state = self.controller.run_cycle_live(goal, input_text)

            events, rewards = self.game_trainer.train(
                game,
                self.game_player,
                self.controller.agents.values()
            )

            stats = self.metrics.compute(events, rewards)

            balance_report = self._safe_parse(
                self.balancer.analyze(game, stats)
            )

            game = self.applier.apply(game, balance_report)

            step += 1

        return {
            "status": "complete",
            "steps": step
        }

    def _safe_parse(self, raw):

        try:
            import json
            return json.loads(raw)
        except:
            return {"is_unbalanced": False, "fixes": []}

    def _apply_layered_mutations(self, game, mutations):

        # RULES
        if "rules" in mutations:
            if "add_rule" in mutations["rules"]:
                game["rules"].append(mutations["rules"]["add_rule"])

        # STATE
        if "state" in mutations:
            for k, v in mutations["state"].get("adjust_parameter", {}).items():
                game[k] = v

        # TERRITORY
        if "territory" in mutations:
            game.setdefault("territory", {})
            game["territory"]["expanded"] = True

        # CONDITIONS
        if "conditions" in mutations:
            for k, v in mutations["conditions"].get("adjust_parameter", {}).items():
                game[k] = v

        return game