1  # Adapted from: https://halite.io/learn-programming-challenge/tutorials/ml-svm
 2  
 3  import copy
 4  import json
 5  import os
 6  import os.path
 7  import zstd
 8  
 9  import hlt
10  
11  ARBITRARY_ID = -1
12  
13  
14  def parse_replay_file(file_name, player_name):
15      print("Load Replay: " + file_name)
16      with open(file_name, 'rb') as f:
17          data = json.loads(zstd.loads(f.read()))
18  
19      print("Load Basic Information")
20      player = [p for p in data['players'] if p['name'].split(" ")[0] == player_name][0]
21      player_id = int(player['player_id'])
22      my_shipyard = hlt.Shipyard(player_id, ARBITRARY_ID,
23                                 hlt.Position(player['factory_location']['x'], player['factory_location']['y']))
24      other_shipyards = [
25          hlt.Shipyard(p['player_id'], ARBITRARY_ID, hlt.Position(p['factory_location']['x'], p['factory_location']['y']))
26          for p in data['players'] if int(p['player_id']) != player_id]
27      width = data['production_map']['width']
28      height = data['production_map']['height']
29  
30      print("Load Cell Information")
31      first_cells = []
32      for x in range(len(data['production_map']['grid'])):
33          row = []
34          for y in range(len(data['production_map']['grid'][x])):
35              row += [hlt.MapCell(hlt.Position(x, y), data['production_map']['grid'][x][y]['energy'])]
36          first_cells.append(row)
37      frames = []
38      for f in data['full_frames']:
39          prev_cells = first_cells if len(frames) == 0 else frames[-1]._cells
40          new_cells = copy.deepcopy(prev_cells)
41          for c in f['cells']:
42              new_cells[c['y']][c['x']].halite_amount = c['production']
43          frames.append(hlt.GameMap(new_cells, width, height))
44  
45      print("Load Player Ships")
46      moves = [{} if str(player_id) not in f['moves'] else {m['id']: m['direction'] for m in f['moves'][str(player_id)] if
47                                                            m['type'] == "m"} for f in data['full_frames']]
48      ships = [{} if str(player_id) not in f['entities'] else {
49          int(sid): hlt.Ship(player_id, int(sid), hlt.Position(ship['x'], ship['y']), ship['energy']) for sid, ship in
50          f['entities'][str(player_id)].items()} for f in data['full_frames']]
51  
52      print("Load Other Player Ships")
53      other_ships = [
54          {int(sid): hlt.Ship(int(pid), int(sid), hlt.Position(ship['x'], ship['y']), ship['energy']) for pid, p in
55           f['entities'].items() if
56           int(pid) != player_id for sid, ship in p.items()} for f in data['full_frames']]
57  
58      print("Load Droppoff Information")
59      first_my_dropoffs = [my_shipyard]
60      first_them_dropoffs = other_shipyards
61      my_dropoffs = []
62      them_dropoffs = []
63      for f in data['full_frames']:
64          new_my_dropoffs = copy.deepcopy(first_my_dropoffs if len(my_dropoffs) == 0 else my_dropoffs[-1])
65          new_them_dropoffs = copy.deepcopy(first_them_dropoffs if len(them_dropoffs) == 0 else them_dropoffs[-1])
66          for e in f['events']:
67              if e['type'] == 'construct':
68                  if int(e['owner_id']) == player_id:
69                      new_my_dropoffs.append(
70                          hlt.Dropoff(player_id, ARBITRARY_ID, hlt.Position(e['location']['x'], e['location']['y'])))
71                  else:
72                      new_them_dropoffs.append(
73                          hlt.Dropoff(e['owner_id'], ARBITRARY_ID, hlt.Position(e['location']['x'], e['location']['y'])))
74          my_dropoffs.append(new_my_dropoffs)
75          them_dropoffs.append(new_them_dropoffs)
76      return list(zip(frames, moves, ships, other_ships, my_dropoffs, them_dropoffs))
77  
78  
79  def parse_replay_folder(folder_name, player_name, max_files=None):
80      replay_buffer = []
81      for file_name in sorted(os.listdir(folder_name)):
82          if not file_name.endswith(".hlt"):
83              continue
84          elif max_files is not None and len(replay_buffer) >= max_files:
85              break
86          else:
87              replay_buffer.append(parse_replay_file(os.path.join(folder_name, file_name), player_name))
88      return replay_buffer