Analysis + Fix for Validating

.gitignore

@@ -159,3 +159,7 @@ Thumbs.db
 
 # vscode
 .vscode/
+
+# Ignore data
+outputs/
+*.png

pyproject.toml

@@ -146,3 +146,6 @@ isort.required-imports = ["from __future__ import annotations"]
 
 [project.scripts]
 p2lab = "P2Lab.__main__:main"
+
+[tool.setuptools]
+package_dir = ["src"]

src/P2Lab/__main__.py

@@ -0,0 +1,37 @@
+"""
+TODO: Write some docs here.
+"""
+
+import asyncio
+from .teams.builder import Builder
+from .evaluator.poke_env import PokeEnv
+from .teams.team import Team 
+from .stats.team_recorder import TeamRecorder
+from tqdm import tqdm
+
+N_generations = 50 # Number of generations to run
+N_teams = 2 # Number of teams to generate per generation
+N_battles = 3 # Number of battles to run per team
+RECORD=True
+
+async def main_loop():
+    builder = Builder(N_seed_teams=N_teams)
+    builder.build_N_teams_from_poke_pool(N_teams)
+    curr_gen = 0 # Current generation
+    evaluator = PokeEnv(n_battles=N_battles)
+    recorder = TeamRecorder()
+
+    # Main expected loop
+    print("Starting main loop and running on Generation: ")
+    for _ in tqdm(range(N_generations)):
+        if RECORD:
+            recorder.record_teams(builder.get_teams(), curr_gen) 
+        await evaluator.evaluate_teams(builder.get_teams())
+        builder.generate_new_teams()
+        curr_gen += 1
+
+def main():
+    asyncio.get_event_loop().run_until_complete(main_loop())
+
+if __name__ == "__main__":
+    main()

src/P2Lab/evaluator/poke_env.py

@@ -0,0 +1,33 @@
+"""
+"""
+import socket
+import numpy as np 
+from itertools import combinations
+from poke_env.player import Player, RandomPlayer
+from poke_env import PlayerConfiguration
+from tqdm import tqdm
+
+SHOW_MATCH_PROGRESS = True
+
+class PokeEnv:
+    def __init__(self, n_battles=100, battle_format="gen7anythinggoes"):
+        self.p1 = RandomPlayer(PlayerConfiguration("Player 1", None), battle_format=battle_format)
+        self.p2 = RandomPlayer(PlayerConfiguration("Player 2", None), battle_format=battle_format)
+        self.n_battles = n_battles
+        pass
+
+    async def evaluate_teams(self, teams):
+        match_ups = [c for c in combinations(np.arange(len(teams)), 2)]
+
+
+        for t1, t2 in ( tqdm(match_ups) if SHOW_MATCH_PROGRESS else match_ups ):
+            res1, res2 = await self.battle(teams[t1], teams[t2])
+            teams[t1].set_fitness(res1)
+            teams[t2].set_fitness(res2)
+
+
+    async def battle(self, team1, team2):
+        self.p1.update_team(team1.to_packed_str())
+        self.p2.update_team(team2.to_packed_str())
+        await self.p1.battle_against(self.p2, n_battles=self.n_battles)
+        return self.p1.n_won_battles, self.p2.n_won_battles

src/p2lab/pokemon/battle.py → src/P2Lab/pokemon/battle.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
-
+from poke_env.player import Player, RandomPlayer
 import numpy as np
 
 

src/P2Lab/stats/__init__.py

@@ -0,0 +1 @@
+""""""

src/P2Lab/stats/team_recorder.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+
+
+class TeamRecorder():
+    def __init__(self) -> None:
+        pass
+
+    def create_folder(self, f):
+        Path(f"outputs/{f}").mkdir(parents=True, exist_ok=True)
+
+    def record(self, team, generation, uid):
+        team.to_df().to_csv(f"outputs/{generation}/{uid}.csv", index=False)
+
+    def record_teams(self, teams, generation):
+        self.create_folder(generation)
+        for idx, team in enumerate(teams):
+            self.record(team, generation, idx)

src/P2Lab/teams/builder.py

@@ -0,0 +1,95 @@
+"""
+The team builder
+"""
+from __future__ import annotations
+
+import sys
+from subprocess import check_output
+import copy
+from tqdm import tqdm
+import numpy as np
+from poke_env.teambuilder import Teambuilder
+from .team import Team
+
+class Builder(Teambuilder):
+    """
+    The team builder
+    """
+    def __init__(self, N_seed_teams=2, teams=None, format="gen7anythinggoes"):
+        if teams:
+            self.teams = [
+                self.join_team(self.parse_showdown_team(team)) for team in teams
+            ]
+        else:
+            self.teams = []
+            print("Generating seed teams")
+            for _ in tqdm(range(N_seed_teams)):
+                poss_team = check_output(f"pokemon-showdown generate-team {format}", shell=True)
+                try:
+                    check_output(f"pokemon-showdown validate-team {format} ", shell=True,input=poss_team)
+                except Exception as e:
+                    print("Error validating team... skipping to next")
+                    continue
+                n_team = self.parse_showdown_team(check_output("pokemon-showdown export-team ",input=poss_team, shell=True).decode(sys.stdout.encoding))
+                if len(n_team)!=6:
+                    raise Exception("Team not of length 6")
+                self.teams.append(n_team)                 
+        self.poke_pool = np.array(self.teams).flatten()
+
+    def build_N_teams_from_poke_pool(self, N_teams):
+        self.teams = [Team(np.random.choice(self.poke_pool, size=6, replace=False)) for _ in range(N_teams)]
+
+    def get_teams(self):
+        return self.teams
+
+    def yield_team(self):
+        return self.teams[np.random.choice(len(self.teams))]
+
+    def generate_new_teams(self):
+        self.last_generation =  copy.deepcopy(self.teams)
+        self.breed_teams()
+        # TODO: Make this fancier
+        #self.mutate_teams()
+
+
+    def breed_teams(self):
+        teams = []
+        for i in range(len(self.teams)):
+            probs = np.array([t.get_fitness() for t in self.teams])
+            probs = probs/np.sum(probs)
+            t1, t2 = np.random.choice(self.teams, 2, p=probs, replace=False)
+            t1 = t1.get_teamlist()
+            t2 = t2.get_teamlist()
+            teams.append(self.breed_team(t1,t2))
+        self.teams = teams
+        self.duplicate_detect() # Prob should take out or move this somewhere else 
+
+    def breed_team(self, t1,t2):
+        # REALLY BAD DUPLICATION PREVENTION
+        # TODO: FIX ASAP      
+        poss_pokes = np.concatenate((t1,t2))
+        names = [str(p).split("|")[0] for p in poss_pokes]
+        new_team_names = np.random.choice(list(set(names)), size=6, replace=False)
+        n2p = {n:p for n,p in zip(names,poss_pokes)}
+        new_team =[n2p[n] for n in new_team_names]
+        return Team(new_team)
+
+    def duplicate_detect(self):
+        for t in self.teams:
+            pokes = []
+            for p in t.get_teamlist():
+                pokes.append(str(p).split("|")[0])
+            if len(set(pokes)) != 6:
+                print("WARNING DUPLICATION DETECTED")
+                print(pokes)
+
+    def mutate_team(self, team, mutation_rate=0.1):
+        # Each pokemon in a team has a given probability of being replaced by a random pokemon from the pool
+        for pokemon in team:
+            if np.random.random() < mutation_rate:
+                new_pokemon = np.random.choice(self.poke_pool)
+                # If the pokemon is already in the team, replace it with a random pokemon from the pool
+                while new_pokemon in team:
+                    new_pokemon = np.random.choice(self.poke_pool)
+                # Replace the pokemon
+                team[np.where(team == pokemon)[0][0]] = new_pokemon

src/P2Lab/teams/team.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+import pandas as pd
+import re
+
+class Team:
+    def __init__(self, pokemon) -> None:
+        self.pokemon = pokemon
+        self.fitness = 0
+
+    def get_teamlist(self):
+        return self.pokemon
+
+    def get_fitness(self):
+        return self.fitness
+
+    def set_fitness(self, x):
+        self.fitness = x
+
+    def to_packed_str(self):
+        return "]".join([mon.formatted for mon in self.pokemon])
+
+    def to_df(self):
+        # TODO Put in the columns or something
+        df = pd.DataFrame([re.split('\||,',s) for s in  re.split('\]', self.to_packed_str())])
+        return df