{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"## This Notebook based \nhttps://www.kaggle.com/code/edwardcrookenden/kore-starter-6th-in-beta-rule-based-agent \nhttps://www.kaggle.com/code/solverworld/kore-beta-1st-place-solution \nhttps://www.kaggle.com/code/lesamu/reinforcement-learning-baseline-in-python \n \nThank you for sharing! \nWithout these I couldn't do anything. \n\n## Approach\nBase agent is Beta 1st place and I added beta 6th offensive commands and commands to expand the base, and made the selection of those commands RL. \nThe opponent operates according to the Beta 1st rule, and is an Agent that randomly executes the Beta 6th command.\n\n## Result\nBeta 6th place ~ 700 \nBeta 1st place ~ 830 \nRL Agent ~ 900 \n \nIt didn't work well. I should set better rules.","metadata":{}},{"cell_type":"code","source":"!pip install --target=lib --no-deps stable-baselines3 gym","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2022-07-14T01:42:55.665493Z","iopub.execute_input":"2022-07-14T01:42:55.666292Z","iopub.status.idle":"2022-07-14T01:43:16.623206Z","shell.execute_reply.started":"2022-07-14T01:42:55.666195Z","shell.execute_reply":"2022-07-14T01:43:16.622167Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile config.py\nimport numpy as np\nfrom kaggle_environments import make\n\n# Read env specification\nENV_SPECIFICATION = make('kore_fleets').specification\nSHIP_COST = ENV_SPECIFICATION.configuration.spawnCost.default\nSHIPYARD_COST = ENV_SPECIFICATION.configuration.convertCost.default\nGAME_CONFIG = {\n 'episodeSteps': ENV_SPECIFICATION.configuration.episodeSteps.default, # You might want to start with smaller values\n 'size': ENV_SPECIFICATION.configuration.size.default,\n 'maxLogLength': None\n}\n\n# Define your opponent. We'll use the starter bot in the notebook environment for this baseline.\nOPPONENT = 'opponent.py'\nGAME_AGENTS = [None, OPPONENT]\n\n# Define our parameters\nN_FEATURES = 4\nMAX_SIPYARD = 20\nACTION_SIZE = (4,)\nDTYPE = np.float64\nMAX_OBSERVABLE_KORE = 500\nMAX_OBSERVABLE_SHIPS = 200\nMIN_ACTION_FLEET_SIZE = 3\nMAX_ACTION_FLEET_SIZE = 246\nMAX_KORE_IN_RESERVE = 40000\nWIN_REWARD = 10000","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.858158Z","iopub.execute_input":"2022-07-14T01:43:16.859166Z","iopub.status.idle":"2022-07-14T01:43:16.878473Z","shell.execute_reply.started":"2022-07-14T01:43:16.859120Z","shell.execute_reply":"2022-07-14T01:43:16.877503Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import os\nimport sys\nKAGGLE_AGENT_PATH = \"/kaggle_simulations/agent/\"\nif os.path.exists(KAGGLE_AGENT_PATH):\n sys.path.insert(0, os.path.join(KAGGLE_AGENT_PATH, 'lib'))\nelse:\n sys.path.insert(0, os.path.join(os.getcwd(), 'lib'))\n\n","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.625435Z","iopub.execute_input":"2022-07-14T01:43:16.626581Z","iopub.status.idle":"2022-07-14T01:43:16.635199Z","shell.execute_reply.started":"2022-07-14T01:43:16.626524Z","shell.execute_reply":"2022-07-14T01:43:16.633852Z"},"jupyter":{"source_hidden":true},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Beta 1st","metadata":{}},{"cell_type":"code","source":"%%writefile basic_1st.py\nimport math\nfrom typing import Union\n\nfrom kaggle_environments.envs.kore_fleets.helpers import SPAWN_VALUES\n\n\ndef max_ships_to_spawn(turns_controlled: int) -> int:\n for idx, target in enumerate(SPAWN_VALUES):\n if turns_controlled < target:\n return idx + 1\n return len(SPAWN_VALUES) + 1\n\n\ndef max_flight_plan_len_for_ship_count(ship_count: int) -> int:\n return math.floor(2 * math.log(ship_count)) + 1\n\n\ndef min_ship_count_for_flight_plan_len(flight_plan_len: int) -> int:\n return math.ceil(math.exp((flight_plan_len - 1) / 2))\n\n\ndef collection_rate_for_ship_count(ship_count: int) -> float:\n return min(math.log(ship_count) / 20, 0.99)\n\n\ndef create_spawn_ships_command(num_ships: int) -> str:\n return f\"SPAWN_{num_ships}\"\n\n\ndef create_launch_fleet_command(num_ships: int, plan: str) -> str:\n return f\"LAUNCH_{num_ships}_{plan}\"\n\n\nclass cached_property:\n \"\"\"\n python 3.9:\n >>> from functools import cached_property\n \"\"\"\n\n def __init__(self, func):\n self.func = func\n self.key = \"__\" + func.__name__\n\n def __get__(self, instance, owner):\n try:\n return instance.__getattribute__(self.key)\n except AttributeError:\n value = self.func(instance)\n instance.__setattr__(self.key, value)\n return value\n\n\nclass cached_call:\n \"\"\"\n may cause a memory leak, be careful\n \"\"\"\n\n def __init__(self, func):\n self.func = func\n self.key = \"__\" + func.__name__\n\n def __get__(self, instance, owner):\n try:\n d = instance.__getattribute__(self.key)\n except AttributeError:\n d = {}\n instance.__setattr__(self.key, d)\n\n def func(x):\n try:\n return d[x]\n except KeyError:\n value = self.func(instance, x)\n d[x] = value\n return value\n\n return func\n\n\nclass Obj:\n def __init__(self, game_id: Union[str, int]):\n self._game_id = game_id\n\n def __repr__(self):\n return f\"{self.__class__.__name__}(id={self._game_id})\"\n\n @property\n def game_id(self):\n return self._game_id","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.637053Z","iopub.execute_input":"2022-07-14T01:43:16.638283Z","iopub.status.idle":"2022-07-14T01:43:16.653684Z","shell.execute_reply.started":"2022-07-14T01:43:16.638235Z","shell.execute_reply":"2022-07-14T01:43:16.652953Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile board_1st.py\nimport itertools\nimport numpy as np\nfrom typing import Dict, List, Union, Optional, Generator\nfrom collections import defaultdict\nfrom kaggle_environments.envs.kore_fleets.helpers import Configuration\n\n\n# <--->\nfrom basic_1st import (\n Obj,\n collection_rate_for_ship_count,\n max_ships_to_spawn,\n cached_property,\n create_spawn_ships_command,\n create_launch_fleet_command,\n)\nfrom geometry_1st import (\n Field,\n Action,\n Point,\n North,\n South,\n Convert,\n PlanPath,\n PlanRoute,\n GAME_ID_TO_ACTION,\n)\nfrom logger_1st import logger\n\n# <--->\n\n\nclass _ShipyardAction:\n def to_str(self):\n raise NotImplementedError\n\n def __repr__(self):\n return self.to_str()\n\n\nclass Spawn(_ShipyardAction):\n def __init__(self, ship_count: int):\n self.ship_count = ship_count\n\n def to_str(self):\n return create_spawn_ships_command(self.ship_count)\n\n\nclass Launch(_ShipyardAction):\n def __init__(self, ship_count: int, route: \"BoardRoute\"):\n self.ship_count = ship_count\n self.route = route\n\n def to_str(self):\n return create_launch_fleet_command(self.ship_count, self.route.plan.to_str())\n\n\nclass DoNothing(_ShipyardAction):\n def __repr__(self):\n return \"Do nothing\"\n\n def to_str(self):\n raise NotImplementedError\n\n\nclass BoardPath:\n max_length = 32\n\n def __init__(self, start: \"Point\", plan: PlanPath):\n assert plan.num_steps > 0 or plan.direction == Convert\n\n self._plan = plan\n\n field = start.field\n x, y = start.x, start.y\n if np.isfinite(plan.num_steps):\n n = plan.num_steps + 1\n else:\n n = self.max_length\n action = plan.direction\n\n if plan.direction == Convert:\n self._track = []\n self._start = start\n self._end = start\n self._build_shipyard = True\n return\n\n if action in (North, South):\n track = field.get_column(x, start=y, size=n * action.dy)\n else:\n track = field.get_row(y, start=x, size=n * action.dx)\n\n self._track = track[1:]\n self._start = start\n self._end = track[-1]\n self._build_shipyard = False\n\n def __repr__(self):\n start, end = self.start, self.end\n return f\"({start.x}, {start.y}) -> ({end.x}, {end.y})\"\n\n def __len__(self):\n return len(self._track)\n\n @property\n def plan(self):\n return self._plan\n\n @property\n def points(self):\n return self._track\n\n @property\n def start(self):\n return self._start\n\n @property\n def end(self):\n return self._end\n\n\nclass BoardRoute:\n def __init__(self, start: \"Point\", plan: \"PlanRoute\"):\n paths = []\n for p in plan.paths:\n path = BoardPath(start, p)\n start = path.end\n paths.append(path)\n\n self._plan = plan\n self._paths = paths\n self._start = paths[0].start\n self._end = paths[-1].end\n\n def __repr__(self):\n points = []\n for p in self._paths:\n points.append(p.start)\n points.append(self.end)\n return \" -> \".join([f\"({p.x}, {p.y})\" for p in points])\n\n def __iter__(self) -> Generator[\"Point\", None, None]:\n for p in self._paths:\n yield from p.points\n\n def __len__(self):\n return sum(len(x) for x in self._paths)\n\n def points(self) -> List[\"Point\"]:\n points = []\n for p in self._paths:\n points += p.points\n return points\n\n @property\n def plan(self) -> PlanRoute:\n return self._plan\n\n def command(self) -> str:\n return self.plan.to_str()\n\n @property\n def paths(self) -> List[BoardPath]:\n return self._paths\n\n @property\n def start(self) -> \"Point\":\n return self._start\n\n @property\n def end(self) -> \"Point\":\n return self._end\n\n def command_length(self) -> int:\n return len(self.command())\n\n def last_action(self):\n return self.paths[-1].plan.direction\n\n def expected_kore(self, board: \"Board\", ship_count: int):\n rate = collection_rate_for_ship_count(ship_count)\n if rate <= 0:\n return 0\n\n point_to_time = {}\n point_to_kore = {}\n for t, p in enumerate(self):\n point_to_time[p] = t\n point_to_kore[p] = p.kore\n\n for f in board.fleets:\n for t, p in enumerate(f.route):\n if p in point_to_time and t < point_to_time[p]:\n point_to_kore[p] *= f.collection_rate\n\n return sum([kore * rate for kore in point_to_kore.values()])\n\n\nclass PositionObj(Obj):\n def __init__(self, *args, point: Point, player_id: int, board: \"Board\", **kwargs):\n super().__init__(*args, **kwargs)\n self._point = point\n self._player_id = player_id\n self._board = board\n\n def __repr__(self):\n return f\"{self.__class__.__name__}(id={self._game_id}, position={self._point}, player={self._player_id})\"\n\n def dirs_to(self, obj: Union[\"PositionObj\", Point]):\n if isinstance(obj, Point):\n return self._point.dirs_to(obj)\n return self._point.dirs_to(obj.point)\n\n def distance_from(self, obj: Union[\"PositionObj\", Point]) -> int:\n if isinstance(obj, Point):\n return self._point.distance_from(obj)\n return self._point.distance_from(obj.point)\n\n @property\n def board(self) -> \"Board\":\n return self._board\n\n @property\n def point(self) -> Point:\n return self._point\n\n @property\n def player_id(self):\n return self._player_id\n\n @property\n def player(self) -> \"Player\":\n return self.board.get_player(self.player_id)\n\n\nclass Shipyard(PositionObj):\n def __init__(self, *args, ship_count: int, turns_controlled: int, **kwargs):\n super().__init__(*args, **kwargs)\n self._ship_count = ship_count\n self._turns_controlled = turns_controlled\n self._guard_ship_count = 0\n self.action: Optional[_ShipyardAction] = None\n\n @property\n def turns_controlled(self):\n return self._turns_controlled\n\n @property\n def max_ships_to_spawn(self) -> int:\n return max_ships_to_spawn(self._turns_controlled)\n\n @property\n def ship_count(self):\n return self._ship_count\n\n @property\n def available_ship_count(self):\n return self._ship_count - self._guard_ship_count\n\n @property\n def guard_ship_count(self):\n return self._guard_ship_count\n\n def set_guard_ship_count(self, ship_count):\n assert ship_count <= self._ship_count\n self._guard_ship_count = ship_count\n\n @cached_property\n def incoming_allied_fleets(self) -> List[\"Fleet\"]:\n fleets = []\n for f in self.board.fleets:\n if f.player_id == self.player_id and f.route.end == self.point:\n fleets.append(f)\n return fleets\n\n @cached_property\n def incoming_hostile_fleets(self) -> List[\"Fleet\"]:\n fleets = []\n for f in self.board.fleets:\n if f.player_id != self.player_id and f.route.end == self.point:\n fleets.append(f)\n return fleets\n\n\nclass Fleet(PositionObj):\n def __init__(\n self,\n *args,\n ship_count: int,\n kore: int,\n route: BoardRoute,\n direction: Action,\n **kwargs,\n ):\n assert ship_count > 0\n assert kore >= 0\n\n super().__init__(*args, **kwargs)\n\n self._ship_count = ship_count\n self._kore = kore\n self._direction = direction\n self._route = route\n\n def __gt__(self, other):\n if self.ship_count != other.ship_count:\n return self.ship_count > other.ship_count\n if self.kore != other.kore:\n return self.kore > other.kore\n return self.direction.game_id > other.direction.game_id\n\n def __lt__(self, other):\n return other.__gt__(self)\n\n @property\n def ship_count(self):\n return self._ship_count\n\n @property\n def kore(self):\n return self._kore\n\n @property\n def route(self):\n return self._route\n\n @property\n def eta(self):\n return len(self._route)\n\n def set_route(self, route: BoardRoute):\n self._route = route\n\n @property\n def direction(self):\n return self._direction\n\n @property\n def collection_rate(self) -> float:\n return collection_rate_for_ship_count(self._ship_count)\n\n def expected_kore(self):\n return self._kore + self._route.expected_kore(self._board, self._ship_count)\n\n def cost(self):\n return self.board.spawn_cost * self.ship_count\n\n def value(self):\n return self.kore / self.cost()\n\n def expected_value(self):\n return self.expected_kore() / self.cost()\n\n\nclass FleetPointer:\n def __init__(self, fleet: Fleet):\n self.obj = fleet\n self.point = fleet.point\n self.is_active = True\n self._paths = []\n self._points = self.points()\n\n def points(self):\n for path in self.obj.route.paths:\n self._paths.append([path.plan.direction, 0])\n for point in path.points:\n self._paths[-1][1] += 1\n yield point\n\n def update(self):\n if not self.is_active:\n self.point = None\n return\n try:\n self.point = next(self._points)\n except StopIteration:\n self.point = None\n self.is_active = False\n\n def current_route(self):\n plan = PlanRoute([PlanPath(d, n) for d, n in self._paths])\n return BoardRoute(self.obj.point, plan)\n\n\nclass Player(Obj):\n def __init__(self, *args, kore: float, board: \"Board\", **kwargs):\n super().__init__(*args, **kwargs)\n self._kore = kore\n self._board = board\n\n @property\n def kore(self):\n return self._kore\n\n def fleet_kore(self):\n return sum(x.kore for x in self.fleets)\n\n def fleet_expected_kore(self):\n return sum(x.expected_kore() for x in self.fleets)\n\n def is_active(self):\n return len(self.fleets) > 0 or len(self.shipyards) > 0\n\n @property\n def board(self):\n return self._board\n\n def _get_objects(self, name):\n d = []\n for x in self._board.__getattribute__(name):\n if x.player_id == self.game_id:\n d.append(x)\n return d\n\n @cached_property\n def fleets(self) -> List[Fleet]:\n return self._get_objects(\"fleets\")\n\n @cached_property\n def shipyards(self) -> List[Shipyard]:\n return self._get_objects(\"shipyards\")\n\n @cached_property\n def ship_count(self) -> int:\n return sum(x.ship_count for x in itertools.chain(self.fleets, self.shipyards))\n\n @cached_property\n def opponents(self) -> List[\"Player\"]:\n return [x for x in self.board.players if x != self]\n\n @cached_property\n def expected_fleets_positions(self) -> Dict[int, Dict[Point, int]]:\n \"\"\"\n time -> point -> fleet\n \"\"\"\n time_to_fleet_positions = defaultdict(dict)\n for f in self.fleets:\n for time, point in enumerate(f.route):\n time_to_fleet_positions[time][point] = f\n return time_to_fleet_positions\n\n @cached_property\n def expected_dmg_positions(self) -> Dict[int, Dict[Point, int]]:\n \"\"\"\n time -> point -> dmg\n \"\"\"\n time_to_dmg_positions = defaultdict(dict)\n for f in self.fleets:\n for time, point in enumerate(f.route):\n for adjacent_point in point.adjacent_points:\n point_to_dmg = time_to_dmg_positions[time]\n if adjacent_point not in point_to_dmg:\n point_to_dmg[adjacent_point] = 0\n point_to_dmg[adjacent_point] += f.ship_count\n return time_to_dmg_positions\n\n def actions(self):\n if self.available_kore() < 0:\n logger.warning(\"Negative balance. Some ships will not spawn.\")\n\n shipyard_id_to_action = {}\n for sy in self.shipyards:\n if not sy.action or isinstance(sy.action, DoNothing):\n continue\n\n shipyard_id_to_action[sy.game_id] = sy.action.to_str()\n return shipyard_id_to_action\n\n def spawn_ship_count(self):\n return sum(\n x.action.ship_count for x in self.shipyards if isinstance(x.action, Spawn)\n )\n\n def need_kore_for_spawn(self):\n return self.board.spawn_cost * self.spawn_ship_count()\n\n def available_kore(self):\n return self._kore - self.need_kore_for_spawn()\n\n\n_FIELD = None\n\n\nclass Board:\n def __init__(self, obs, conf):\n self._conf = Configuration(conf)\n self._step = obs[\"step\"]\n\n global _FIELD\n if _FIELD is None or self._step == 0:\n _FIELD = Field(self._conf.size)\n else:\n assert _FIELD.size == self._conf.size\n\n self._field: Field = _FIELD\n\n id_to_point = {x.game_id: x for x in self._field}\n\n for point_id, kore in enumerate(obs[\"kore\"]):\n point = id_to_point[point_id]\n point.set_kore(kore)\n\n self._players = []\n self._fleets = []\n self._shipyards = []\n for player_id, player_data in enumerate(obs[\"players\"]):\n player_kore, player_shipyards, player_fleets = player_data\n player = Player(game_id=player_id, kore=player_kore, board=self)\n self._players.append(player)\n\n for fleet_id, fleet_data in player_fleets.items():\n point_id, kore, ship_count, direction, flight_plan = fleet_data\n position = id_to_point[point_id]\n direction = GAME_ID_TO_ACTION[direction]\n if ship_count < self.shipyard_cost and Convert.command in flight_plan:\n # can't convert\n flight_plan = \"\".join(\n [x for x in flight_plan if x != Convert.command]\n )\n plan = PlanRoute.from_str(flight_plan, direction)\n route = BoardRoute(position, plan)\n fleet = Fleet(\n game_id=fleet_id,\n point=position,\n player_id=player_id,\n ship_count=ship_count,\n kore=kore,\n route=route,\n direction=direction,\n board=self,\n )\n self._fleets.append(fleet)\n\n for shipyard_id, shipyard_data in player_shipyards.items():\n point_id, ship_count, turns_controlled = shipyard_data\n position = id_to_point[point_id]\n shipyard = Shipyard(\n game_id=shipyard_id,\n point=position,\n player_id=player_id,\n ship_count=ship_count,\n turns_controlled=turns_controlled,\n board=self,\n )\n self._shipyards.append(shipyard)\n\n self._players = [x for x in self._players if x.is_active()]\n\n self._update_fleets_destination()\n\n def __getitem__(self, item):\n return self._field[item]\n\n def __iter__(self):\n return self._field.__iter__()\n\n @property\n def field(self):\n return self._field\n\n @property\n def size(self):\n return self._field.size\n\n @property\n def step(self):\n return self._step\n\n @property\n def steps_left(self):\n return self._conf.episode_steps - self._step - 1\n\n @property\n def shipyard_cost(self):\n return self._conf.convert_cost\n\n @property\n def spawn_cost(self):\n return self._conf.spawn_cost\n\n @property\n def regen_rate(self):\n return self._conf.regen_rate\n\n @property\n def max_cell_kore(self):\n return self._conf.max_cell_kore\n\n @property\n def players(self) -> List[Player]:\n return self._players\n\n @property\n def fleets(self) -> List[Fleet]:\n return self._fleets\n\n @property\n def shipyards(self) -> List[Shipyard]:\n return self._shipyards\n\n def get_player(self, game_id) -> Player:\n for p in self._players:\n if p.game_id == game_id:\n return p\n raise KeyError(f\"Player `{game_id}` doas not exists.\")\n\n def get_obj_at_point(self, point: Point) -> Optional[Union[Fleet, Shipyard]]:\n for x in itertools.chain(self.fleets, self.shipyards):\n if x.point == point:\n return x\n\n def _update_fleets_destination(self):\n \"\"\"\n trying to predict future positions\n very inaccurate\n \"\"\"\n\n shipyard_positions = {x.point for x in self.shipyards}\n\n fleets = [FleetPointer(f) for f in self.fleets]\n\n while any(x.is_active for x in fleets):\n for f in fleets:\n f.update()\n\n # fleet to shipyard\n for f in fleets:\n if f.point in shipyard_positions:\n f.is_active = False\n\n # allied fleets\n for player in self.players:\n point_to_fleets = defaultdict(list)\n for f in fleets:\n if f.is_active and f.obj.player_id == player.game_id:\n point_to_fleets[f.point].append(f)\n for point_fleets in point_to_fleets.values():\n if len(point_fleets) > 1:\n for f in sorted(point_fleets, key=lambda x: x.obj)[:-1]:\n f.is_active = False\n\n # fleet to fleet\n point_to_fleets = defaultdict(list)\n for f in fleets:\n if f.is_active:\n point_to_fleets[f.point].append(f)\n for point_fleets in point_to_fleets.values():\n if len(point_fleets) > 1:\n for f in sorted(point_fleets, key=lambda x: x.obj)[:-1]:\n f.is_active = False\n\n # adjacent damage\n point_to_fleet = {}\n for f in fleets:\n if f.is_active:\n point_to_fleet[f.point] = f\n\n point_to_dmg = defaultdict(int)\n for point, fleet in point_to_fleet.items():\n for p in point.adjacent_points:\n if p in point_to_fleet:\n adjacent_fleet = point_to_fleet[p]\n if adjacent_fleet.obj.player_id != fleet.obj.player_id:\n point_to_dmg[p] += fleet.obj.ship_count\n\n for point, fleet in point_to_fleet.items():\n dmg = point_to_dmg[point]\n if fleet.obj.ship_count <= dmg:\n fleet.is_active = False\n\n for f in fleets:\n f.obj.set_route(f.current_route())","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.655944Z","iopub.execute_input":"2022-07-14T01:43:16.656471Z","iopub.status.idle":"2022-07-14T01:43:16.680573Z","shell.execute_reply.started":"2022-07-14T01:43:16.656433Z","shell.execute_reply":"2022-07-14T01:43:16.679629Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile control_1st.py\n\nimport random\n\n# <--->\nfrom geometry_1st import PlanRoute\nfrom board_1st import Player, Launch, Spawn, Fleet, FleetPointer, BoardRoute\nfrom helpers_1st import is_invitable_victory, find_shortcut_routes\nfrom logger_1st import logger\n\n# <--->\n\n\ndef direct_attack(agent: Player, max_distance: int = 10):\n board = agent.board\n\n max_distance = min(board.steps_left, max_distance)\n\n targets = []\n for x in agent.opponents:\n for sy in x.shipyards:\n for fleet in sy.incoming_allied_fleets:\n if fleet.expected_value() > 0.5:\n targets.append(fleet)\n\n if not targets:\n return\n\n shipyards = [\n x for x in agent.shipyards if x.available_ship_count > 0 and not x.action\n ]\n if not shipyards:\n return\n\n point_to_closest_shipyard = {}\n for p in board:\n closest_shipyard = None\n min_distance = board.size\n for sy in agent.shipyards:\n distance = sy.point.distance_from(p)\n if distance < min_distance:\n min_distance = distance\n closest_shipyard = sy\n point_to_closest_shipyard[p] = closest_shipyard.point\n\n opponent_shipyard_points = {x.point for x in board.shipyards if x.player_id != agent.game_id}\n for t in targets:\n min_ships_to_send = int(t.ship_count * 1.2)\n attacked = False\n\n for sy in shipyards:\n if sy.action or sy.available_ship_count < min_ships_to_send:\n continue\n\n num_ships_to_launch = sy.available_ship_count\n\n for target_time, target_point in enumerate(t.route, 1):\n if target_time > max_distance:\n continue\n\n if sy.point.distance_from(target_point) != target_time:\n continue\n\n paths = sy.point.dirs_to(target_point)\n random.shuffle(paths)\n plan = PlanRoute(paths)\n destination = point_to_closest_shipyard[target_point]\n\n paths = target_point.dirs_to(destination)\n random.shuffle(paths)\n plan += PlanRoute(paths)\n if num_ships_to_launch < plan.min_fleet_size():\n continue\n\n route = BoardRoute(sy.point, plan)\n\n if any(x in opponent_shipyard_points for x in route.points()):\n continue\n\n if is_intercept_direct_attack_route(route, agent, direct_attack_fleet=t):\n continue\n\n logger.info(\n f\"Direct attack {sy.point}->{target_point}, distance={target_time}\"\n )\n sy.action = Launch(num_ships_to_launch, route)\n attacked = True\n break\n\n if attacked:\n break\n\n\ndef is_intercept_direct_attack_route(\n route: BoardRoute, player: Player, direct_attack_fleet: Fleet\n):\n board = player.board\n\n fleets = [FleetPointer(f) for f in board.fleets if f != direct_attack_fleet]\n\n for point in route.points()[:-1]:\n for fleet in fleets:\n fleet.update()\n\n if fleet.point is None:\n continue\n\n if fleet.point == point:\n return True\n\n if fleet.obj.player_id != player.game_id:\n for p in fleet.point.adjacent_points:\n if p == point:\n return True\n\n return False\n\n\ndef adjacent_attack(agent: Player, max_distance: int = 10):\n board = agent.board\n\n max_distance = min(board.steps_left, max_distance)\n\n targets = _find_adjacent_targets(agent, max_distance)\n if not targets:\n return\n\n shipyards = [\n x for x in agent.shipyards if x.available_ship_count > 0 and not x.action\n ]\n if not shipyards:\n return\n\n fleets_to_be_attacked = set()\n for t in sorted(targets, key=lambda x: (-len(x[\"fleets\"]), x[\"time\"])):\n target_point = t[\"point\"]\n target_time = t[\"time\"]\n target_fleets = t[\"fleets\"]\n if any(x in fleets_to_be_attacked for x in target_fleets):\n continue\n\n for sy in shipyards:\n if sy.action:\n continue\n\n distance = sy.distance_from(target_point)\n if distance > target_time:\n continue\n min_ship_count = min(x.ship_count for x in target_fleets)\n num_ships_to_send = min(sy.available_ship_count, min_ship_count)\n\n routes = find_shortcut_routes(\n board,\n sy.point,\n target_point,\n agent,\n num_ships_to_send,\n route_distance=target_time,\n )\n if not routes:\n continue\n\n route = random.choice(routes)\n logger.info(\n f\"Adjacent attack {sy.point}->{target_point}, distance={distance}, target_time={target_time}\"\n )\n sy.action = Launch(num_ships_to_send, route)\n for fleet in target_fleets:\n fleets_to_be_attacked.add(fleet)\n break\n\n\ndef _find_adjacent_targets(agent: Player, max_distance: int = 5):\n board = agent.board\n shipyards_points = {x.point for x in board.shipyards}\n fleets = [FleetPointer(f) for f in board.fleets]\n if len(fleets) < 2:\n return []\n\n time = 0\n targets = []\n while any(x.is_active for x in fleets) and time <= max_distance:\n time += 1\n\n for f in fleets:\n f.update()\n\n point_to_fleet = {\n x.point: x.obj\n for x in fleets\n if x.is_active and x.point not in shipyards_points\n }\n\n for point in board:\n if point in point_to_fleet or point in shipyards_points:\n continue\n\n adjacent_fleets = [\n point_to_fleet[x] for x in point.adjacent_points if x in point_to_fleet\n ]\n if len(adjacent_fleets) < 2:\n continue\n\n if any(x.player_id == agent.game_id for x in adjacent_fleets):\n continue\n\n targets.append({\"point\": point, \"time\": time, \"fleets\": adjacent_fleets})\n\n return targets\n\n\ndef _need_more_ships(agent: Player, ship_count: int):\n board = agent.board\n if board.steps_left < 10:\n return False\n if ship_count > _max_ships_to_control(agent):\n return False\n if board.steps_left < 50 and is_invitable_victory(agent):\n return False\n return True\n\n\ndef _max_ships_to_control(agent: Player):\n return max(100, 3 * sum(x.ship_count for x in agent.opponents))\n\n\ndef greedy_spawn(agent: Player):\n board = agent.board\n\n if not _need_more_ships(agent, agent.ship_count):\n return\n\n ship_count = agent.ship_count\n max_ship_count = _max_ships_to_control(agent)\n for shipyard in agent.shipyards:\n if shipyard.action:\n continue\n\n if shipyard.ship_count > agent.ship_count * 0.2 / len(agent.shipyards):\n continue\n\n num_ships_to_spawn = shipyard.max_ships_to_spawn\n if int(agent.available_kore() // board.spawn_cost) >= num_ships_to_spawn:\n shipyard.action = Spawn(num_ships_to_spawn)\n\n ship_count += num_ships_to_spawn\n if ship_count > max_ship_count:\n return\n\n\ndef spawn(agent: Player):\n board = agent.board\n\n if not _need_more_ships(agent, agent.ship_count):\n return\n\n ship_count = agent.ship_count\n max_ship_count = _max_ships_to_control(agent)\n for shipyard in agent.shipyards:\n if shipyard.action:\n continue\n num_ships_to_spawn = min(\n int(agent.available_kore() // board.spawn_cost),\n shipyard.max_ships_to_spawn,\n )\n if num_ships_to_spawn:\n shipyard.action = Spawn(num_ships_to_spawn)\n ship_count += num_ships_to_spawn\n if ship_count > max_ship_count:\n return\n","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.682253Z","iopub.execute_input":"2022-07-14T01:43:16.682782Z","iopub.status.idle":"2022-07-14T01:43:16.702070Z","shell.execute_reply.started":"2022-07-14T01:43:16.682747Z","shell.execute_reply":"2022-07-14T01:43:16.701128Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile defence_1st.py\n\nimport random\n\n# <--->\nfrom board_1st import Spawn, Player, Launch\nfrom helpers_1st import find_shortcut_routes\nfrom logger_1st import logger\n\n# <--->\n\n\ndef defend_shipyards(agent: Player):\n board = agent.board\n\n need_help_shipyards = []\n for sy in agent.shipyards:\n if sy.action:\n continue\n\n incoming_hostile_fleets = sy.incoming_hostile_fleets\n incoming_allied_fleets = sy.incoming_allied_fleets\n\n if not incoming_hostile_fleets:\n continue\n\n incoming_hostile_power = sum(x.ship_count for x in incoming_hostile_fleets)\n incoming_hostile_time = min(x.eta for x in incoming_hostile_fleets)\n incoming_allied_power = sum(\n x.ship_count\n for x in incoming_allied_fleets\n if x.eta < incoming_hostile_time\n )\n\n ships_needed = incoming_hostile_power - incoming_allied_power\n if sy.ship_count > ships_needed:\n sy.set_guard_ship_count(min(sy.ship_count, int(ships_needed * 1.1)))\n continue\n\n # spawn as much as possible\n num_ships_to_spawn = min(\n int(agent.available_kore() // board.spawn_cost), sy.max_ships_to_spawn\n )\n if num_ships_to_spawn:\n logger.debug(f\"Spawn ships to protect shipyard {sy.point}\")\n sy.action = Spawn(num_ships_to_spawn)\n\n need_help_shipyards.append(sy)\n\n for sy in need_help_shipyards:\n incoming_hostile_fleets = sy.incoming_hostile_fleets\n incoming_hostile_time = min(x.eta for x in incoming_hostile_fleets)\n\n for other_sy in agent.shipyards:\n if other_sy == sy or other_sy.action or not other_sy.available_ship_count:\n continue\n\n distance = other_sy.distance_from(sy)\n if distance == incoming_hostile_time - 1:\n routes = find_shortcut_routes(\n board, other_sy.point, sy.point, agent, other_sy.ship_count\n )\n if routes:\n logger.info(f\"Send reinforcements {other_sy.point}->{sy.point}\")\n other_sy.action = Launch(\n other_sy.available_ship_count, random.choice(routes)\n )\n elif distance < incoming_hostile_time - 1:\n other_sy.set_guard_ship_count(other_sy.ship_count)","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.703870Z","iopub.execute_input":"2022-07-14T01:43:16.704352Z","iopub.status.idle":"2022-07-14T01:43:16.719504Z","shell.execute_reply.started":"2022-07-14T01:43:16.704308Z","shell.execute_reply":"2022-07-14T01:43:16.718739Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile expantion_1st.py\nimport random\nfrom typing import List\nfrom collections import defaultdict\n\n# <--->\nfrom basic_1st import min_ship_count_for_flight_plan_len\nfrom geometry_1st import Point, Convert, PlanRoute, PlanPath\nfrom board_1st import Player, BoardRoute, Launch\n\n# <--->\n\n\ndef expand(player: Player):\n board = player.board\n num_shipyards_to_create = need_more_shipyards(player)\n if not num_shipyards_to_create:\n return\n\n shipyard_positions = {x.point for x in board.shipyards}\n\n shipyard_to_point = find_best_position_for_shipyards(player)\n\n shipyard_count = 0\n for shipyard, target in shipyard_to_point.items():\n if shipyard_count >= num_shipyards_to_create:\n break\n\n if shipyard.available_ship_count < board.shipyard_cost or shipyard.action:\n continue\n\n incoming_hostile_fleets = shipyard.incoming_hostile_fleets\n if incoming_hostile_fleets:\n continue\n\n target_distance = shipyard.distance_from(target)\n\n routes = []\n for p in board:\n if p in shipyard_positions:\n continue\n\n distance = shipyard.distance_from(p) + p.distance_from(target)\n if distance > target_distance:\n continue\n\n plan = PlanRoute(shipyard.dirs_to(p) + p.dirs_to(target))\n route = BoardRoute(shipyard.point, plan)\n\n if shipyard.available_ship_count < min_ship_count_for_flight_plan_len(\n len(route.plan.to_str()) + 1\n ):\n continue\n\n route_points = route.points()\n if any(x in shipyard_positions for x in route_points):\n continue\n\n if not is_safety_route_to_convert(route_points, player):\n continue\n\n routes.append(route)\n\n if routes:\n route = random.choice(routes)\n route = BoardRoute(\n shipyard.point, route.plan + PlanRoute([PlanPath(Convert)])\n )\n shipyard.action = Launch(shipyard.available_ship_count, route)\n shipyard_count += 1\n\n\ndef find_best_position_for_shipyards(player: Player):\n board = player.board\n shipyards = board.shipyards\n\n shipyard_to_scores = defaultdict(list)\n for p in board:\n if p.kore > 50:\n continue\n\n closed_shipyard = None\n min_distance = board.size\n for shipyard in shipyards:\n distance = shipyard.point.distance_from(p)\n if shipyard.player_id != player.game_id:\n distance -= 1\n\n if distance < min_distance:\n closed_shipyard = shipyard\n min_distance = distance\n\n if (\n not closed_shipyard\n or closed_shipyard.player_id != player.game_id\n or min_distance < 3\n or min_distance > 5\n ):\n continue\n\n nearby_kore = sum(x.kore for x in p.nearby_points(10))\n nearby_shipyards = sum(1 for x in board.shipyards if x.distance_from(p) < 5)\n score = nearby_kore - 1000 * nearby_shipyards - 1000 * min_distance\n shipyard_to_scores[closed_shipyard].append({\"score\": score, \"point\": p})\n\n shipyard_to_point = {}\n for shipyard, scores in shipyard_to_scores.items():\n if scores:\n scores = sorted(scores, key=lambda x: x[\"score\"])\n point = scores[-1][\"point\"]\n shipyard_to_point[shipyard] = point\n\n return shipyard_to_point\n\n\ndef need_more_shipyards(player: Player) -> int:\n board = player.board\n\n if player.ship_count < 100:\n return 0\n\n fleet_distance = []\n for sy in player.shipyards:\n for f in sy.incoming_allied_fleets:\n fleet_distance.append(len(f.route))\n\n if not fleet_distance:\n return 0\n\n mean_fleet_distance = sum(fleet_distance) / len(fleet_distance)\n\n shipyard_production_capacity = sum(x.max_ships_to_spawn for x in player.shipyards)\n\n steps_left = board.steps_left\n if steps_left > 100:\n scale = 3\n elif steps_left > 50:\n scale = 4\n elif steps_left > 10:\n scale = 100\n else:\n scale = 1000\n\n needed = player.kore > scale * shipyard_production_capacity * mean_fleet_distance\n if not needed:\n return 0\n\n current_shipyard_count = len(player.shipyards)\n\n op_shipyard_positions = {\n x.point for x in board.shipyards if x.player_id != player.game_id\n }\n expected_shipyard_count = current_shipyard_count + sum(\n 1\n for x in player.fleets\n if x.route.last_action() == Convert or x.route.end in op_shipyard_positions\n )\n\n opponent_shipyard_count = max(len(x.shipyards) for x in player.opponents)\n opponent_ship_count = max(x.ship_count for x in player.opponents)\n if (\n expected_shipyard_count > opponent_shipyard_count\n and player.ship_count < opponent_ship_count\n ):\n return 0\n\n if current_shipyard_count < 10:\n if expected_shipyard_count > current_shipyard_count:\n return 0\n else:\n return 1\n\n return max(0, 5 - (expected_shipyard_count - current_shipyard_count))\n\n\ndef is_safety_route_to_convert(route_points: List[Point], player: Player):\n board = player.board\n\n target_point = route_points[-1]\n target_time = len(route_points)\n for pl in board.players:\n if pl != player:\n for t, positions in pl.expected_fleets_positions.items():\n if t >= target_time and target_point in positions:\n return False\n\n shipyard_positions = {x.point for x in board.shipyards}\n\n for time, point in enumerate(route_points):\n for pl in board.players:\n if point in shipyard_positions:\n return False\n\n is_enemy = pl != player\n\n if point in pl.expected_fleets_positions[time]:\n return False\n\n if is_enemy:\n if point in pl.expected_dmg_positions[time]:\n return False\n\n return True ","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.721392Z","iopub.execute_input":"2022-07-14T01:43:16.722212Z","iopub.status.idle":"2022-07-14T01:43:16.739104Z","shell.execute_reply.started":"2022-07-14T01:43:16.722166Z","shell.execute_reply":"2022-07-14T01:43:16.738168Z"},"_kg_hide-input":true,"jupyter":{"source_hidden":true},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile geometry_1st.py\nimport numpy as np\nfrom typing import Tuple, List, Generator\n\n# <--->\nfrom basic_1st import Obj, cached_call, cached_property, min_ship_count_for_flight_plan_len\n\n# <--->\n\n\nclass Action(Obj):\n def __init__(self, dx, dy, game_id, command):\n super().__init__(game_id)\n self._dx = dx\n self._dy = dy\n self._command = command\n\n def __repr__(self):\n return self._command\n\n @property\n def dx(self) -> int:\n return self._dx\n\n @property\n def dy(self) -> int:\n return self._dy\n\n @property\n def command(self) -> str:\n return self._command\n\n\nNorth = Action(\n dx=0,\n dy=1,\n game_id=0,\n command=\"N\",\n)\nEast = Action(\n dx=1,\n dy=0,\n game_id=1,\n command=\"E\",\n)\nSouth = Action(\n dx=0,\n dy=-1,\n command=\"S\",\n game_id=2,\n)\nWest = Action(\n dx=-1,\n dy=0,\n command=\"W\",\n game_id=3,\n)\nConvert = Action(\n dx=0,\n dy=0,\n command=\"C\",\n game_id=-1,\n)\n\n\nALL_DIRECTIONS = {North, East, South, West}\nALL_ACTIONS = {North, East, South, West, Convert}\nGAME_ID_TO_ACTION = {x.game_id: x for x in ALL_ACTIONS}\nCOMMAND_TO_ACTION = {x.command: x for x in ALL_ACTIONS}\nACTION_TO_OPPOSITE_ACTION = {\n North: South,\n East: West,\n South: North,\n West: East,\n}\n\n\ndef get_opposite_action(action):\n return ACTION_TO_OPPOSITE_ACTION.get(action, action)\n\n\nclass Point(Obj):\n def __init__(self, x: int, y: int, kore: float, field: \"Field\"):\n super().__init__(game_id=(field.size - y - 1) * field.size + x)\n self._x = x\n self._y = y\n self._kore = kore\n self._field = field\n\n def __repr__(self):\n return f\"Point({self._x}, {self._y})\"\n\n @property\n def x(self) -> int:\n return self._x\n\n @property\n def y(self) -> int:\n return self._y\n\n def to_tuple(self) -> Tuple[int, int]:\n return self._x, self._y\n\n @property\n def kore(self) -> float:\n return self._kore\n\n def set_kore(self, kore: float):\n self._kore = kore\n\n @property\n def field(self) -> \"Field\":\n return self._field\n\n def apply(self, action: Action) -> \"Point\":\n return self._field[(self.x + action.dx, self.y + action.dy)]\n\n @cached_call\n def distance_from(self, point: \"Point\") -> int:\n return sum(p.num_steps for p in self.dirs_to(point))\n\n @cached_property\n def adjacent_points(self) -> List[\"Point\"]:\n return [self.apply(a) for a in ALL_DIRECTIONS]\n\n @cached_property\n def row(self) -> List[\"Point\"]:\n return list(self._field.points[:, self.y])\n\n @cached_property\n def column(self) -> List[\"Point\"]:\n return list(self._field.points[self.x, :])\n\n @cached_call\n def nearby_points(self, r: int) -> List[\"Point\"]:\n if r > 1:\n points = []\n for p in self._field:\n distance = self.distance_from(p)\n if 0 < distance <= r:\n points.append(p)\n return points\n elif r == 1:\n return self.adjacent_points\n\n raise ValueError(\"Radius must be more or equal then 1\")\n\n @cached_call\n def dirs_to(self, point: \"Point\") -> List[\"PlanPath\"]:\n dx, dy = self._field.swap(self._x - point.x, self._y - point.y)\n ret = []\n if dx:\n ret.append(PlanPath(West, dx))\n if dy:\n ret.append(PlanPath(South, dy))\n return ret\n\n\nclass Field:\n def __init__(self, size: int):\n self._size = size\n self._points = self.create_array(size)\n\n def __iter__(self) -> Generator[Point, None, None]:\n for row in self._points:\n yield from row\n\n def create_array(self, size: int) -> np.ndarray:\n ar = np.zeros((size, size), dtype=Point)\n for x in range(size):\n for y in range(size):\n point = Point(x, y, kore=0, field=self)\n ar[x, y] = point\n return ar\n\n @property\n def points(self) -> np.ndarray:\n return self._points\n\n def get_row(self, y: int, start: int, size: int) -> List[Point]:\n if size < 0:\n return self.get_row(y, start=start + size + 1, size=-size)[::-1]\n\n ps = self._points\n start %= self._size\n out = []\n while size > 0:\n d = list(ps[slice(start, start + size), y])\n size -= len(d)\n start = 0\n out += d\n return out\n\n def get_column(self, x: int, start: int, size: int) -> List[Point]:\n if size < 0:\n return self.get_column(x, start=start + size + 1, size=-size)[::-1]\n\n ps = self._points\n start %= self._size\n out = []\n while size > 0:\n d = list(ps[x, slice(start, start + size)])\n size -= len(d)\n start = 0\n out += d\n return out\n\n @property\n def size(self) -> int:\n return self._size\n\n def swap(self, dx, dy):\n size = self._size\n if abs(dx) > size / 2:\n dx -= np.sign(dx) * size\n if abs(dy) > size / 2:\n dy -= np.sign(dy) * size\n return dx, dy\n\n def __getitem__(self, item) -> Point:\n x, y = item\n return self._points[x % self._size, y % self._size]\n\n\nclass PlanPath:\n def __init__(self, direction: Action, num_steps: int = 0):\n if direction == Convert:\n self._direction = direction\n self._num_steps = 0\n elif num_steps > 0:\n self._direction = direction\n self._num_steps = num_steps\n else:\n self._direction = get_opposite_action(direction)\n self._num_steps = -num_steps\n\n def __repr__(self):\n return self.to_str()\n\n @property\n def direction(self):\n return self._direction\n\n @property\n def num_steps(self):\n return self._num_steps\n\n def to_str(self):\n if self.direction == Convert:\n return Convert.command\n elif self.num_steps == 0:\n return \"\"\n elif self.num_steps == 1:\n return self.direction.command\n else:\n return self.direction.command + str(self.num_steps - 1)\n\n def reverse(self) -> \"PlanPath\":\n return PlanPath(self.direction, -self.num_steps)\n\n\nclass PlanRoute:\n def __init__(self, paths: List[PlanPath]):\n self._paths = self.simplify(paths)\n\n def __repr__(self):\n return self.to_str()\n\n def __add__(self, other: \"PlanRoute\") -> \"PlanRoute\":\n return PlanRoute(self.paths + other.paths)\n\n def __bool__(self):\n return bool(self._paths)\n\n @property\n def paths(self):\n return self._paths\n\n @property\n def num_steps(self):\n return sum(x.num_steps for x in self._paths)\n\n @classmethod\n def simplify(cls, paths: List[PlanPath]):\n if not paths:\n return paths\n\n new_paths = []\n last_path = None\n for p in paths:\n if last_path and p.direction == last_path.direction:\n new_paths[-1] = PlanPath(p.direction, p.num_steps + last_path.num_steps)\n else:\n last_path = p\n new_paths.append(p)\n return new_paths\n\n def command_length(self):\n return len(self.to_str())\n\n def min_fleet_size(self):\n return min_ship_count_for_flight_plan_len(self.command_length())\n\n def reverse(self) -> \"PlanRoute\":\n return PlanRoute([x.reverse() for x in self.paths])\n\n @property\n def actions(self):\n actions = []\n for p in self.paths:\n actions += [p.direction for _ in range(p.num_steps)]\n return actions\n\n @classmethod\n def from_str(cls, str_plan: str, current_direction: Action) -> \"PlanRoute\":\n if current_direction not in ALL_DIRECTIONS:\n raise ValueError(f\"Unknown direction `{current_direction}`\")\n\n if not str_plan:\n return PlanRoute([PlanPath(current_direction, np.inf)])\n\n commands = []\n for x in str_plan:\n if x in COMMAND_TO_ACTION:\n commands.append([])\n commands[-1].append(x)\n elif x.isdigit():\n if not commands:\n commands = [[]]\n commands[-1].append(x)\n else:\n raise ValueError(f\"Unknown command `{x}`.\")\n\n paths = []\n for i, p in enumerate(commands):\n if i == 0 and p[0].isdigit():\n action = current_direction\n num_steps = int(\"\".join(p))\n if num_steps == 0:\n continue\n else:\n action = COMMAND_TO_ACTION[p[0]]\n if len(p) == 1:\n num_steps = 1\n else:\n num_steps = int(\"\".join(p[1:])) + 1\n\n paths.append(PlanPath(direction=action, num_steps=num_steps))\n if action == Convert:\n break\n\n if not paths:\n return PlanRoute([PlanPath(current_direction, np.inf)])\n\n last_direction = paths[-1].direction\n if last_direction != Convert:\n paths[-1] = PlanPath(direction=last_direction, num_steps=np.inf)\n\n return PlanRoute(paths)\n\n def to_str(self) -> str:\n s = \"\"\n for a in self.paths[:-1]:\n s += a.to_str()\n s += self.paths[-1].direction.command\n return s","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.740395Z","iopub.execute_input":"2022-07-14T01:43:16.741108Z","iopub.status.idle":"2022-07-14T01:43:16.760681Z","shell.execute_reply.started":"2022-07-14T01:43:16.741064Z","shell.execute_reply":"2022-07-14T01:43:16.759785Z"},"_kg_hide-input":true,"jupyter":{"source_hidden":true},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile helpers_1st.py\nimport random\nfrom typing import List\n\n\n# <--->\nfrom geometry_1st import Point\nfrom board_1st import Board, Player, BoardRoute, PlanRoute\n\n# <--->\n\n\ndef is_intercept_route(\n route: BoardRoute, player: Player, safety=True, allow_shipyard_intercept=False\n):\n board = player.board\n\n if not allow_shipyard_intercept:\n shipyard_points = {x.point for x in board.shipyards}\n else:\n shipyard_points = {}\n\n for time, point in enumerate(route.points()[:-1]):\n if point in shipyard_points:\n return True\n\n for pl in board.players:\n is_enemy = pl != player\n\n if point in pl.expected_fleets_positions[time]:\n return True\n\n if safety and is_enemy:\n if point in pl.expected_dmg_positions[time]:\n return True\n\n return False\n\n\ndef find_shortcut_routes(\n board: Board,\n start: Point,\n end: Point,\n player: Player,\n num_ships: int,\n safety: bool = True,\n allow_shipyard_intercept=False,\n route_distance=None\n) -> List[BoardRoute]:\n if route_distance is None:\n route_distance = start.distance_from(end)\n routes = []\n for p in board:\n distance = start.distance_from(p) + p.distance_from(end)\n if distance != route_distance:\n continue\n\n path1 = start.dirs_to(p)\n path2 = p.dirs_to(end)\n random.shuffle(path1)\n random.shuffle(path2)\n\n plan = PlanRoute(path1 + path2)\n\n if num_ships < plan.min_fleet_size():\n continue\n\n route = BoardRoute(start, plan)\n\n if is_intercept_route(\n route,\n player,\n safety=safety,\n allow_shipyard_intercept=allow_shipyard_intercept,\n ):\n continue\n\n routes.append(route)\n\n return routes\n\n\ndef is_invitable_victory(player: Player):\n if not player.opponents:\n return True\n\n board = player.board\n if board.steps_left > 100:\n return False\n\n board_kore = sum(x.kore for x in board) * (1 + board.regen_rate) ** board.steps_left\n\n player_kore = player.kore + player.fleet_expected_kore()\n opponent_kore = max(x.kore + x.fleet_expected_kore() for x in player.opponents)\n return player_kore > opponent_kore + board_kore","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.762245Z","iopub.execute_input":"2022-07-14T01:43:16.762658Z","iopub.status.idle":"2022-07-14T01:43:16.778406Z","shell.execute_reply.started":"2022-07-14T01:43:16.762620Z","shell.execute_reply":"2022-07-14T01:43:16.777259Z"},"_kg_hide-input":true,"jupyter":{"source_hidden":true},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile logger_1st.py\nimport os\nimport logging\n\nFILE = \"game.log\"\nIS_KAGGLE = os.path.exists(\"/kaggle_simulations\")\nLEVEL = logging.DEBUG if not IS_KAGGLE else logging.INFO\nLOGGING_ENABLED = False\n\n\nclass _FileHandler(logging.FileHandler):\n def emit(self, record):\n if not LOGGING_ENABLED:\n return\n\n if IS_KAGGLE:\n print(self.format(record))\n else:\n super().emit(record)\n\n\ndef init_logger(_logger):\n if not IS_KAGGLE:\n if os.path.exists(FILE):\n os.remove(FILE)\n\n while _logger.hasHandlers():\n _logger.removeHandler(_logger.handlers[0])\n\n _logger.setLevel(LEVEL)\n ch = _FileHandler(FILE)\n ch.setLevel(LEVEL)\n formatter = logging.Formatter(\n \"%(asctime)s - %(levelname)s - %(message)s\", datefmt=\"%H-%M-%S\"\n )\n ch.setFormatter(formatter)\n _logger.addHandler(ch)\n\n\nlogger = logging.getLogger()","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.782827Z","iopub.execute_input":"2022-07-14T01:43:16.783547Z","iopub.status.idle":"2022-07-14T01:43:16.797936Z","shell.execute_reply.started":"2022-07-14T01:43:16.783491Z","shell.execute_reply":"2022-07-14T01:43:16.797229Z"},"_kg_hide-input":true,"jupyter":{"source_hidden":true},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile opponent_1st.py\nimport random\n# <--->\nfrom board_1st import Board\nfrom logger_1st import logger, init_logger\nfrom offence_1st import capture_shipyards\nfrom defence_1st import defend_shipyards\nfrom expantion_1st import expand\nfrom mining_1st import mine\nfrom control_1st import spawn, greedy_spawn, adjacent_attack, direct_attack\n\n# <--->\n\n\ndef agent(obs, conf):\n if obs[\"step\"] == 0:\n init_logger(logger)\n\n board = Board(obs, conf)\n step = board.step\n my_id = obs[\"player\"]\n remaining_time = obs[\"remainingOverageTime\"]\n logger.info(f\", remaining_time={remaining_time:.1f}\")\n\n try:\n a = board.get_player(my_id)\n except KeyError:\n return {}\n\n if not a.opponents:\n return {}\n r = random.choice(range(2))\n if r ==0:\n defend_shipyards(a)\n capture_shipyards(a)\n adjacent_attack(a)\n direct_attack(a)\n expand(a)\n greedy_spawn(a)\n mine(a)\n spawn(a)\n elif r == 1:\n capture_shipyards(a)\n adjacent_attack(a)\n direct_attack(a)\n defend_shipyards(a)\n expand(a)\n greedy_spawn(a)\n mine(a)\n spawn(a)\n return a.actions()","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.799250Z","iopub.execute_input":"2022-07-14T01:43:16.799683Z","iopub.status.idle":"2022-07-14T01:43:16.815510Z","shell.execute_reply.started":"2022-07-14T01:43:16.799652Z","shell.execute_reply":"2022-07-14T01:43:16.814303Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile mining_1st.py\nimport random\nimport numpy as np\nfrom typing import List\nfrom collections import defaultdict\n\n# <--->\nfrom geometry_1st import PlanRoute\nfrom board_1st import Player, BoardRoute, Launch, Shipyard\nfrom helpers_1st import is_intercept_route\n\n# <--->\n\n\ndef mine(agent: Player):\n board = agent.board\n if not agent.opponents:\n return\n\n safety = False\n my_ship_count = agent.ship_count\n op_ship_count = max(x.ship_count for x in agent.opponents)\n if my_ship_count < 2 * op_ship_count:\n safety = True\n\n op_ship_count = []\n for op in agent.opponents:\n for fleet in op.fleets:\n op_ship_count.append(fleet.ship_count)\n\n if not op_ship_count:\n mean_fleet_size = 0\n max_fleet_size = np.inf\n else:\n mean_fleet_size = np.percentile(op_ship_count, 75)\n max_fleet_size = int(max(op_ship_count) * 1.1)\n\n point_to_score = estimate_board_risk(agent)\n\n shipyard_count = len(agent.shipyards)\n if shipyard_count < 10:\n max_distance = 15\n elif shipyard_count < 20:\n max_distance = 12\n else:\n max_distance = 8\n\n max_distance = min(int(board.steps_left // 2), max_distance)\n\n for sy in agent.shipyards:\n if sy.action:\n continue\n\n free_ships = sy.available_ship_count\n\n if free_ships <= 2:\n continue\n\n routes = find_shipyard_mining_routes(\n sy, safety=safety, max_distance=max_distance\n )\n\n route_to_score = {}\n for route in routes:\n route_points = route.points()\n\n if all(point_to_score[x] > 0 for x in route_points):\n num_ships_to_launch = free_ships\n else:\n if free_ships < mean_fleet_size:\n continue\n num_ships_to_launch = min(free_ships, max_fleet_size)\n\n score = route.expected_kore(board, num_ships_to_launch) / len(route)\n route_to_score[route] = score\n\n if not route_to_score:\n continue\n\n routes = sorted(route_to_score, key=lambda x: -route_to_score[x])\n for route in routes:\n if all(point_to_score[x] >= 1 for x in route):\n num_ships_to_launch = free_ships\n else:\n num_ships_to_launch = min(free_ships, 199)\n if num_ships_to_launch < route.plan.min_fleet_size():\n continue\n else:\n sy.action = Launch(num_ships_to_launch, route)\n break\n\n\ndef estimate_board_risk(player: Player):\n board = player.board\n\n shipyard_to_area = defaultdict(list)\n for p in board:\n closest_shipyard = None\n min_distance = board.size\n for sh in board.shipyards:\n distance = sh.point.distance_from(p)\n if distance < min_distance:\n closest_shipyard = sh\n min_distance = distance\n\n shipyard_to_area[closest_shipyard].append(p)\n\n point_to_score = {}\n for sy, points in shipyard_to_area.items():\n if sy.player_id == player.game_id:\n for p in points:\n point_to_score[p] = 1\n else:\n for p in points:\n point_to_score[p] = -1\n\n return point_to_score\n\n\ndef find_shipyard_mining_routes(\n sy: Shipyard, safety=True, max_distance: int = 15\n) -> List[BoardRoute]:\n if max_distance < 1:\n return []\n\n departure = sy.point\n player = sy.player\n\n destinations = set()\n for shipyard in sy.player.shipyards:\n siege = sum(x.ship_count for x in shipyard.incoming_hostile_fleets)\n if siege >= shipyard.ship_count:\n continue\n destinations.add(shipyard.point)\n\n if not destinations:\n return []\n\n routes = []\n for c in sy.point.nearby_points(max_distance):\n if c == departure or c in destinations:\n continue\n\n paths = departure.dirs_to(c)\n random.shuffle(paths)\n plan = PlanRoute(paths)\n destination = sorted(destinations, key=lambda x: c.distance_from(x))[0]\n if destination == departure:\n plan += plan.reverse()\n else:\n paths = c.dirs_to(destination)\n random.shuffle(paths)\n plan += PlanRoute(paths)\n\n route = BoardRoute(departure, plan)\n\n if is_intercept_route(route, player, safety):\n continue\n\n routes.append(BoardRoute(departure, plan))\n\n return routes","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.817450Z","iopub.execute_input":"2022-07-14T01:43:16.818131Z","iopub.status.idle":"2022-07-14T01:43:16.835864Z","shell.execute_reply.started":"2022-07-14T01:43:16.818085Z","shell.execute_reply":"2022-07-14T01:43:16.835156Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile offence_1st.py\nimport random\n\nimport numpy as np\nfrom collections import defaultdict\n\n# <--->\nfrom basic_1st import max_ships_to_spawn\nfrom board_1st import Player, Shipyard, Launch\nfrom helpers_1st import find_shortcut_routes\nfrom logger_1st import logger\n\n# <--->\n\n\nclass _ShipyardTarget:\n def __init__(self, shipyard: Shipyard):\n self.shipyard = shipyard\n self.point = shipyard.point\n self.expected_profit = self._estimate_profit()\n self.reinforcement_distance = self._get_reinforcement_distance()\n self._future_ship_count = self._estimate_future_ship_count()\n self.total_incoming_power = self._get_total_incoming_power()\n\n def __repr__(self):\n return f\"Target {self.shipyard}\"\n\n def estimate_shipyard_power(self, time):\n return self._future_ship_count[time]\n\n def _get_total_incoming_power(self):\n return sum(x.ship_count for x in self.shipyard.incoming_allied_fleets)\n\n def _get_reinforcement_distance(self):\n incoming_allied_fleets = self.shipyard.incoming_allied_fleets\n if not incoming_allied_fleets:\n return np.inf\n return min(x.eta for x in incoming_allied_fleets)\n\n def _estimate_profit(self):\n board = self.shipyard.board\n spawn_cost = board.spawn_cost\n profit = sum(\n 2 * x.expected_kore() - x.ship_count * spawn_cost\n for x in self.shipyard.incoming_allied_fleets\n )\n profit += spawn_cost * board.shipyard_cost\n return profit\n\n def _estimate_future_ship_count(self):\n shipyard = self.shipyard\n player = shipyard.player\n board = shipyard.board\n\n time_to_fleet_kore = defaultdict(int)\n for sh in player.shipyards:\n for f in sh.incoming_allied_fleets:\n time_to_fleet_kore[len(f.route)] += f.expected_kore()\n\n shipyard_reinforcements = defaultdict(int)\n for f in shipyard.incoming_allied_fleets:\n shipyard_reinforcements[len(f.route)] += f.ship_count\n\n spawn_cost = board.spawn_cost\n player_kore = player.kore\n ship_count = shipyard.ship_count\n future_ship_count = [ship_count]\n for t in range(1, board.size + 1):\n ship_count += shipyard_reinforcements[t]\n player_kore += time_to_fleet_kore[t]\n\n can_spawn = max_ships_to_spawn(shipyard.turns_controlled + t)\n spawn_count = min(int(player_kore // spawn_cost), can_spawn)\n player_kore -= spawn_count * spawn_cost\n ship_count += spawn_count\n future_ship_count.append(ship_count)\n\n return future_ship_count\n\n\ndef capture_shipyards(agent: Player, max_attack_distance=10):\n board = agent.board\n agent_shipyards = [\n x for x in agent.shipyards if x.available_ship_count >= 3 and not x.action\n ]\n if not agent_shipyards:\n return\n\n targets = []\n for op_sy in board.shipyards:\n if op_sy.player_id == agent.game_id or op_sy.incoming_hostile_fleets:\n continue\n target = _ShipyardTarget(op_sy)\n # if target.expected_profit > 0:\n targets.append(target)\n\n if not targets:\n return\n\n for t in targets:\n shipyards = sorted(\n agent_shipyards, key=lambda x: t.point.distance_from(x.point)\n )\n\n for sy in shipyards:\n if sy.action:\n continue\n\n distance = sy.point.distance_from(t.point)\n if distance > max_attack_distance:\n continue\n\n power = t.estimate_shipyard_power(distance)\n\n if sy.available_ship_count <= power:\n continue\n\n num_ships_to_launch = min(sy.available_ship_count, int(power * 1.2))\n\n routes = find_shortcut_routes(\n board,\n sy.point,\n t.point,\n agent,\n num_ships_to_launch,\n )\n if routes:\n route = random.choice(routes)\n logger.info(\n f\"Attack shipyard {sy.point}->{t.point}\"\n )\n sy.action = Launch(num_ships_to_launch, route)\n break","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.837010Z","iopub.execute_input":"2022-07-14T01:43:16.837474Z","iopub.status.idle":"2022-07-14T01:43:16.856833Z","shell.execute_reply.started":"2022-07-14T01:43:16.837443Z","shell.execute_reply":"2022-07-14T01:43:16.855824Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Beta 6th","metadata":{}},{"cell_type":"code","source":"%%writefile extra_helpers.py\n\nfrom kaggle_environments.envs.kore_fleets.helpers import *\nfrom random import choice, randint, randrange, sample, seed, random\n\ndef get_col_row(size, pos):\n return pos % size, pos // size\n\ndef get_to_pos(size, pos, direction):\n col, row = get_col_row(size, pos)\n if direction == \"NORTH\":\n return pos - size if pos >= size else size ** 2 - size + col\n elif direction == \"SOUTH\":\n return col if pos + size >= size ** 2 else pos + size\n elif direction == \"EAST\":\n return pos + 1 if col < size - 1 else row * size\n elif direction == \"WEST\":\n return pos - 1 if col > 0 else (row + 1) * size - 1\n \ndef get_shortest_flight_path_between(position_a, position_b, size, trailing_digits=False):\n mag_x = 1 if position_b.x > position_a.x else -1\n abs_x = abs(position_b.x - position_a.x)\n dir_x = mag_x if abs_x < size/2 else -mag_x\n mag_y = 1 if position_b.y > position_a.y else -1\n abs_y = abs(position_b.y - position_a.y)\n dir_y = mag_y if abs_y < size/2 else -mag_y\n flight_path_x = \"\"\n if abs_x > 0:\n flight_path_x += \"E\" if dir_x == 1 else \"W\"\n flight_path_x += str(abs_x - 1) if (abs_x - 1) > 0 else \"\"\n flight_path_y = \"\"\n if abs_y > 0:\n flight_path_y += \"N\" if dir_y == 1 else \"S\"\n flight_path_y += str(abs_y - 1) if (abs_y - 1) > 0 else \"\"\n if not len(flight_path_x) == len(flight_path_y):\n if len(flight_path_x) < len(flight_path_y):\n return flight_path_x + (flight_path_y if trailing_digits else flight_path_y[0])\n else:\n return flight_path_y + (flight_path_x if trailing_digits else flight_path_x[0])\n return flight_path_y + (flight_path_x if trailing_digits or not flight_path_x else flight_path_x[0]) if random() < .5 else flight_path_x + (flight_path_y if trailing_digits or not flight_path_y else flight_path_y[0])\n\ndef get_total_ships(board, player):\n ships = 0\n for fleet in board.fleets.values():\n if fleet.player_id == player:\n ships += fleet.ship_count\n for shipyard in board.shipyards.values():\n if shipyard.player_id == player:\n ships += shipyard.ship_count\n return ships \n\n# ref @egrehbbt \ndef max_flight_plan_len_for_ship_count(ship_count):\n return math.floor(2 * math.log(ship_count)) + 1\n\n# ref @egrehbbt \ndef min_ship_count_for_flight_plan_len(flight_plan_len):\n return math.ceil(math.exp((flight_plan_len - 1) / 2))\n\n# ref @egrehbbt \ndef collection_rate_for_ship_count(ship_count):\n return min(math.log(ship_count) / 20, 0.99)\n\ndef spawn_ships(shipyard, remaining_kore, spawn_cost):\n return ShipyardAction.spawn_ships(min(shipyard.max_spawn, int(remaining_kore/spawn_cost)))","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.879915Z","iopub.execute_input":"2022-07-14T01:43:16.880799Z","iopub.status.idle":"2022-07-14T01:43:16.897176Z","shell.execute_reply.started":"2022-07-14T01:43:16.880757Z","shell.execute_reply":"2022-07-14T01:43:16.896269Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile defend.py\n\nfrom kaggle_environments.envs.kore_fleets.helpers import *\n\ndef should_defend(board, me, shipyard, radius=7):\n loc = shipyard.position\n for i in range(1-radius, radius):\n for j in range(1-radius, radius):\n pos = loc.translate(Point(i, j), board.configuration.size)\n if ((board.cells.get(pos).fleet is not None) \n and (board.cells.get(pos).fleet.ship_count > 50)\n and (board.cells.get(pos).fleet.player_id!=me.id)\n and ((board.cells.get(pos).fleet.ship_count) > shipyard.ship_count)):\n return True \n return False","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.898470Z","iopub.execute_input":"2022-07-14T01:43:16.899538Z","iopub.status.idle":"2022-07-14T01:43:16.913052Z","shell.execute_reply.started":"2022-07-14T01:43:16.899483Z","shell.execute_reply":"2022-07-14T01:43:16.912242Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile attack.py\n\nfrom kaggle_environments.envs.kore_fleets.helpers import *\n\ndef should_attack(board, shipyard, remaining_kore, spawn_cost, invading_fleet_size):\n closest_enemy_shipyard = get_closest_enemy_shipyard(board, shipyard.position, board.current_player)\n dist_to_closest_enemy_shipyard = 100 if not closest_enemy_shipyard else shipyard.position.distance_to(closest_enemy_shipyard.position, board.configuration.size)\n if (closest_enemy_shipyard \n and (closest_enemy_shipyard.ship_count < 20 or dist_to_closest_enemy_shipyard < 15) \n and (remaining_kore >= spawn_cost or shipyard.ship_count >= invading_fleet_size) \n and (board.step > 300 or dist_to_closest_enemy_shipyard < 12)):\n return True\n return False\n\ndef get_closest_enemy_shipyard(board, position, me):\n min_dist = 1000000\n enemy_shipyard = None\n for shipyard in board.shipyards.values():\n if shipyard.player_id == me.id:\n continue\n dist = position.distance_to(shipyard.position, board.configuration.size)\n if dist < min_dist:\n min_dist = dist\n enemy_shipyard = shipyard\n return enemy_shipyard\n","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.914320Z","iopub.execute_input":"2022-07-14T01:43:16.914798Z","iopub.status.idle":"2022-07-14T01:43:16.929639Z","shell.execute_reply.started":"2022-07-14T01:43:16.914755Z","shell.execute_reply":"2022-07-14T01:43:16.928669Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile build.py\n\nfrom kaggle_environments.envs.kore_fleets.helpers import *\n\ndef should_build(shipyard, remaining_kore):\n if remaining_kore > 500 and shipyard.max_spawn > 5:\n return True\n return False\n\ndef check_location(board, loc, me):\n if board.cells.get(loc).shipyard and board.cells.get(loc).shipyard.player.id == me.id:\n return 0\n kore = 0\n for i in range(-6, 7):\n for j in range(-6, 7):\n pos = loc.translate(Point(i, j), board.configuration.size)\n kore += board.cells.get(pos).kore or 0\n return kore\n\ndef build_new_shipyard(shipyard, board, me, convert_cost, search_radius=3):\n best_dir = 0\n best_kore = 0\n best_gap1 = 0\n best_gap2 = 0\n for i in range(4):\n next_dir = (i + 1) % 4\n for gap1 in range(0, search_radius, 1):\n for gap2 in range(0, search_radius, 1):\n enemy_shipyard_close = False\n diff1 = Direction.from_index(i).to_point() * gap1\n diff2 = Direction.from_index(next_dir).to_point() * gap2\n diff = diff1 + diff2\n pos = shipyard.position.translate(diff, board.configuration.size)\n for shipyard in board.shipyards.values():\n if ((shipyard.player_id != me.id)\n and (pos.distance_to(shipyard.position, board.configuration.size) < 4)):\n enemy_shipyard_close = True\n if enemy_shipyard_close:\n continue\n h = check_location(board, pos, me)\n if h > best_kore:\n best_kore = h\n best_gap1 = gap1\n best_gap2 = gap2\n best_dir = i\n gap1 = str(best_gap1)\n gap2 = str(best_gap2)\n next_dir = (best_dir + 1) % 4\n flight_plan = Direction.list_directions()[best_dir].to_char() + gap1\n flight_plan += Direction.list_directions()[next_dir].to_char() + gap2\n flight_plan += \"C\"\n return ShipyardAction.launch_fleet_with_flight_plan(max(convert_cost + 30, int(shipyard.ship_count/2)), flight_plan)","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.931396Z","iopub.execute_input":"2022-07-14T01:43:16.931848Z","iopub.status.idle":"2022-07-14T01:43:16.942677Z","shell.execute_reply.started":"2022-07-14T01:43:16.931811Z","shell.execute_reply":"2022-07-14T01:43:16.942011Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile mine.py\n\nfrom kaggle_environments.envs.kore_fleets.helpers import *\nfrom extra_helpers import *\n\ndef should_mine(shipyard, best_fleet_size):\n if shipyard.ship_count >= best_fleet_size:\n return True\n return False\n\ndef check_path(board, start, dirs, dist_a, dist_b, collection_rate, L=False):\n kore = 0\n npv = .99\n current = start\n steps = 2 * (dist_a + dist_b + 2)\n for idx, d in enumerate(dirs):\n if L and idx==2:\n break\n for _ in range((dist_a if idx % 2 == 0 else dist_b) + 1):\n current = current.translate(d.to_point(), board.configuration.size)\n kore += int((board.cells.get(current).kore or 0) * collection_rate)\n final_kore = int((board.cells.get(current).kore or 0) * collection_rate)\n if L: kore = (kore) + (kore*(1-collection_rate)) - final_kore\n return math.pow(npv, steps) * kore / steps\n\ndef get_circular_flight_plan(gap1, gap2, start_dir):\n flight_plan = Direction.list_directions()[start_dir].to_char()\n if int(gap1):\n flight_plan += gap1\n next_dir = (start_dir + 1) % 4\n flight_plan += Direction.list_directions()[next_dir].to_char()\n if int(gap2):\n flight_plan += gap2\n next_dir = (next_dir + 1) % 4\n flight_plan += Direction.list_directions()[next_dir].to_char()\n if int(gap1):\n flight_plan += gap1\n next_dir = (next_dir + 1) % 4\n flight_plan += Direction.list_directions()[next_dir].to_char()\n return flight_plan\n\ndef get_L_flight_plan(gap1, gap2, start_dir):\n flight_plan = Direction.list_directions()[start_dir].to_char()\n if int(gap1):\n flight_plan += gap1\n next_dir = (start_dir + 1) % 4\n flight_plan += Direction.list_directions()[next_dir].to_char()\n if int(gap2):\n flight_plan += gap2\n next_dir = (next_dir + 2) % 4\n flight_plan += Direction.list_directions()[next_dir].to_char()\n if int(gap2):\n flight_plan += gap2\n next_dir = (next_dir - 1) % 4\n flight_plan += Direction.list_directions()[next_dir].to_char()\n return flight_plan\n\ndef get_rectangle_flight_plan(gap, start_dir):\n flight_plan = Direction.list_directions()[start_dir].to_char()\n next_dir = (start_dir + 1) % 4\n flight_plan += Direction.list_directions()[next_dir].to_char()\n if int(gap):\n flight_plan += gap\n next_dir = (next_dir + 1) % 4\n flight_plan += Direction.list_directions()[next_dir].to_char()\n next_dir = (next_dir + 1) % 4\n flight_plan += Direction.list_directions()[next_dir].to_char()\n return flight_plan\n\ndef check_flight_paths(board, shipyard, search_radius):\n best_h = 0\n best_gap1 = 1\n best_gap2 = 1\n best_dir = board.step % 4\n for i in range(4):\n dirs = Direction.list_directions()[i:] + Direction.list_directions()[:i]\n for gap1 in range(0, search_radius):\n for gap2 in range(0, search_radius):\n fleet_size = min_ship_count_for_flight_plan_len(7)\n h = check_path(board, shipyard.position, dirs, gap1, gap2, collection_rate_for_ship_count(fleet_size), L=False)\n if h/fleet_size > best_h:\n best_h = h/fleet_size\n best_flight_plan = get_circular_flight_plan(str(gap1), str(gap2), i)\n best_fleet_size = fleet_size\n h = check_path(board, shipyard.position, dirs, gap1, gap2, collection_rate_for_ship_count(collection_rate_for_ship_count(fleet_size)), L=True)\n if h/fleet_size > best_h:\n best_h = h/fleet_size\n best_flight_plan = get_L_flight_plan(str(gap1), str(gap2), i)\n best_fleet_size = fleet_size\n if gap1!=0:\n continue\n fleet_size = min_ship_count_for_flight_plan_len(5)\n h = check_path(board, shipyard.position, dirs, gap1, gap2, collection_rate_for_ship_count(fleet_size), L=False)\n if h/fleet_size > best_h:\n best_h = h/fleet_size\n best_flight_plan = get_rectangle_flight_plan(str(gap2), i)\n best_fleet_size = fleet_size \n return best_fleet_size, best_flight_plan ","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.944859Z","iopub.execute_input":"2022-07-14T01:43:16.945485Z","iopub.status.idle":"2022-07-14T01:43:16.961733Z","shell.execute_reply.started":"2022-07-14T01:43:16.945408Z","shell.execute_reply":"2022-07-14T01:43:16.961062Z"},"jupyter":{"source_hidden":true},"_kg_hide-input":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile opponent.py\n\nfrom kaggle_environments.envs.kore_fleets.helpers import *\nfrom extra_helpers import *\nfrom defend import *\nfrom attack import *\nfrom build import *\nfrom mine import *\nfrom random import randint\nimport itertools\nimport numpy as np\nfrom random import choice, randint, randrange, sample, seed, random\nimport math\nfrom opponent_1st import agent as agent_1st\n\n# Random Agent based Beta 6th place agent\ndef _agent(obs, config):\n board = Board(obs, config)\n me = board.current_player\n remaining_kore = me.kore\n shipyards = me.shipyards\n convert_cost = board.configuration.convert_cost\n size = board.configuration.size\n spawn_cost = board.configuration.spawn_cost\n turn = board.step\n \n defence_radius = 7\n \n shipyards = sample(shipyards, len(shipyards))\n for shipyard in shipyards:\n invading_fleet_size = randint(50, 155)\n convert_cost_buffer = randint(50, 155)\n mining_search_radius = randint(3, 10)\n best_fleet_size, best_flight_plan = check_flight_paths(board, shipyard, mining_search_radius) \n \n move = -1\n if random() > 0.0:\n # Random move\n move = choice(range(4))\n else:\n # Rule based\n if should_defend(board, me, shipyard, defence_radius):\n move = 0\n elif should_attack(board, shipyard, remaining_kore, spawn_cost, invading_fleet_size):\n move = 1\n elif should_build(shipyard, remaining_kore):\n move = 2\n elif should_mine(shipyard, best_fleet_size):\n move = 3\n \n if move == 0:\n if remaining_kore >= spawn_cost:\n shipyard.next_action = spawn_ships(shipyard, remaining_kore, spawn_cost)\n \n elif move == 1:\n if shipyard.ship_count >= invading_fleet_size:\n closest_enemy_shipyard = get_closest_enemy_shipyard(board, shipyard.position, board.current_player)\n if closest_enemy_shipyard is None:\n if remaining_kore >= spawn_cost:\n shipyard.next_action = spawn_ships(shipyard, remaining_kore, spawn_cost)\n else:\n flight_plan = get_shortest_flight_path_between(shipyard.position, closest_enemy_shipyard.position, size)\n shipyard.next_action = ShipyardAction.launch_fleet_with_flight_plan(invading_fleet_size, flight_plan)\n elif remaining_kore >= spawn_cost:\n shipyard.next_action = spawn_ships(shipyard, remaining_kore, spawn_cost)\n\n elif move == 2:\n if shipyard.ship_count >= convert_cost + convert_cost_buffer:\n shipyard.next_action = build_new_shipyard(shipyard, board, me, convert_cost)\n elif remaining_kore >= spawn_cost:\n shipyard.next_action = spawn_ships(shipyard, remaining_kore, spawn_cost)\n \n elif move == 3:\n shipyard.next_action = ShipyardAction.launch_fleet_with_flight_plan(best_fleet_size, best_flight_plan)\n \n elif (remaining_kore > spawn_cost):\n shipyard.next_action = spawn_ships(shipyard, remaining_kore, spawn_cost)\n \n elif (len(me.fleet_ids) == 0 and shipyard.ship_count <= 22) and len(shipyards)==1:\n if remaining_kore > 11:\n shipyard.next_action = spawn_ships(shipyard, remaining_kore, spawn_cost)\n else:\n direction = Direction.NORTH\n if shipyard.ship_count > 0:\n shipyard.next_action = ShipyardAction.launch_fleet_with_flight_plan(shipyard.ship_count, direction.to_char())\n \n return me.next_actions\n\n\ndef agent(obs, config):\n r = random()\n if r > 0.5:\n return agent_1st(obs, config)\n else:\n return _agent(obs, config)","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.962869Z","iopub.execute_input":"2022-07-14T01:43:16.963349Z","iopub.status.idle":"2022-07-14T01:43:16.982762Z","shell.execute_reply.started":"2022-07-14T01:43:16.963318Z","shell.execute_reply":"2022-07-14T01:43:16.981774Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile reward_utils.py\nfrom config import GAME_CONFIG, SHIP_COST, SHIPYARD_COST\nfrom kaggle_environments.envs.kore_fleets.helpers import Board\nimport numpy as np\nfrom math import floor\n# Compute weight constants -- See get_board_value's docstring\n_max_steps = GAME_CONFIG['episodeSteps']\n_end_of_asset_value = floor(.5 * _max_steps)\n_weights_assets = np.linspace(start=1, stop=0.5, num=_end_of_asset_value)\n_weights_assets2 = np.linspace(start=1, stop=0.5, num=_end_of_asset_value)\n_weights_shipyard = np.linspace(start=0.5, stop=1, num=_end_of_asset_value)\n_weights_kore = np.linspace(start=0, stop=0.1, num=_end_of_asset_value)\n\nWEIGHTS_ASSETS = np.append(_weights_assets, np.zeros(_max_steps - _end_of_asset_value))\nWEIGHTS_SHIPYARD = np.append(_weights_shipyard, np.zeros(_max_steps - _end_of_asset_value))\nWEIGHTS_ASSETS2 = np.append(_weights_assets2, np.zeros(_max_steps - _end_of_asset_value))\nWEIGHTS_KORE = np.append(_weights_kore, np.ones(_max_steps - _end_of_asset_value))\nWEIGHTS_MAX_SPAWN = {x: (x+3)/4 for x in range(1, 11)} # Value multiplier of a shipyard as a function of its max spawn\nWEIGHTS_KORE_IN_FLEETS = WEIGHTS_KORE * WEIGHTS_ASSETS/2 # Always equal or smaller than either, almost always smaller\n\n\ndef get_board_value(board: Board) -> float:\n \"\"\"Computes the board value for the current player.\n\n The board value captures how are we currently performing, compared to the opponent. Each player's partial board\n value assesses the player's situation, taking into account their current kore, ship count, shipyard count\n (including their max spawn) and kore carried by fleets. We then define the board value as the difference between\n player's partial board values.\n Flight plans and the positioning of fleet and shipyards do not flow into the board value (yet).\n\n To keep things simple, we'll take a weighted sum as the partial board value. We need weighting since\n the importance of each item changes over time. We don't need to have the most kore at the beginning of the game,\n but we do at the end. Ship count won't help us win games in the latter stages, but it is crucial in the beginning.\n Fleets and shipyards will be accounted for proportionally to their kore cost.\n\n For efficiency, the weight factors are pre-computed at module level. Here is the logic behind the weighting:\n WEIGHTS_KORE: Applied to the player's kore count. Increases linearly from 0 to 1. It reaches one before\n the maximum game length is reached.\n WEIGHTS_ASSETS: Applied to fleets and shipyards. Decreases linearly from 1 to 0 and reaches zero before the maximum\n length. It emphasizes the need of having ships over kore at the beginning of the game.\n WEIGHTS_MAX_SPAWN: Shipyard value is multiplied by its max spawn. This captures the idea that long-held shipyards\n are more valuable.\n WEIGHTS_KORE_IN_FLEETS: Kore in fleets should be valued, too. But its value must be upper-bounded by WEIGHTS_KORE\n (it can never be better to have kore in cargo than home) and it must decrease in time, since it doesn't\n count towards the end kore count.\n\n Args:\n board: The board for which we want to compute the value.\n\n Returns:\n The value of the board.\n \"\"\"\n board_value = 0\n if not board:\n return board_value\n\n # Get the weights as a function of the current game step\n step = board.step\n weight_shipyard = WEIGHTS_SHIPYARD[step]\n\n # Compute the partial board values\n for player in board.players.values():\n player_fleets, player_shipyards = list(player.fleets), list(player.shipyards)\n\n value_shipyards = weight_shipyard * SHIPYARD_COST * (\n sum(shipyard.max_spawn * WEIGHTS_MAX_SPAWN[shipyard.max_spawn] for shipyard in player_shipyards)\n )\n\n # Add (or subtract) the partial values to the total board value. The current player is always us.\n modifier = 1 if player.is_current_player else -1\n# board_value += modifier * (value_kore + value_fleets + value_shipyards + value_kore_in_cargo)\n board_value += modifier * (value_shipyards)\n\n return board_value","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:16.984357Z","iopub.execute_input":"2022-07-14T01:43:16.984762Z","iopub.status.idle":"2022-07-14T01:43:17.005530Z","shell.execute_reply.started":"2022-07-14T01:43:16.984721Z","shell.execute_reply":"2022-07-14T01:43:17.004742Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile environment.py\nimport gym\nimport numpy as np\nfrom random import randint\nfrom gym import spaces\nfrom math import floor\nfrom kaggle_environments import make\nfrom kaggle_environments.envs.kore_fleets.helpers import ShipyardAction, Board, Direction\nfrom typing import Union, Tuple, Dict\nfrom reward_utils import get_board_value\nfrom config import (\n N_FEATURES,\n ACTION_SIZE,\n GAME_AGENTS,\n GAME_CONFIG,\n DTYPE,\n MAX_OBSERVABLE_KORE,\n MAX_OBSERVABLE_SHIPS,\n MIN_ACTION_FLEET_SIZE,\n MAX_ACTION_FLEET_SIZE,\n MAX_KORE_IN_RESERVE,\n WIN_REWARD,\n MAX_SIPYARD\n)\n\nprint(\n N_FEATURES,\n ACTION_SIZE,\n GAME_AGENTS,\n GAME_CONFIG,\n DTYPE,\n MAX_OBSERVABLE_KORE,\n MAX_OBSERVABLE_SHIPS,\n MAX_ACTION_FLEET_SIZE,\n MAX_KORE_IN_RESERVE,\n WIN_REWARD,)\n\nfrom kaggle_environments.envs.kore_fleets.helpers import *\nfrom extra_helpers import *\nfrom defend import *\nfrom attack import *\nfrom build import *\nfrom mine import *\nfrom random import randint\nimport itertools\nimport numpy as np\nfrom random import choice, randint, randrange, sample, seed, random\nimport math\n\n# <--->\nfrom board_1st import Board as Board_1st\nfrom logger_1st import logger, init_logger\nfrom offence_1st import capture_shipyards\nfrom defence_1st import defend_shipyards\nfrom expantion_1st import expand\nfrom mining_1st import mine\nfrom control_1st import spawn, greedy_spawn, adjacent_attack, direct_attack\n\n# <--->\n\n\nclass KoreGymEnv(gym.Env):\n \"\"\"An openAI-gym env wrapper for kaggle's kore environment. Can be used with stable-baselines3.\n\n There are three fundamental components to this class which you would want to customize for your own agents:\n The action space is defined by `action_space` and `gym_to_kore_action()`\n The state space (observations) is defined by `state_space` and `obs_as_gym_state()`\n The reward is computed with `compute_reward()`\n\n Note that the action and state spaces define the inputs and outputs to your model *as numpy arrays*. Use the\n functions mentioned above to translate these arrays into actual kore environment observations and actions.\n\n The rest is basically boilerplate and makes sure that the kaggle environment plays nicely with stable-baselines3.\n\n Usage:\n >>> from stable_baselines3 import PPO\n >>>\n >>> kore_env = KoreGymEnv()\n >>> model = PPO('MlpPolicy', kore_env, verbose=1)\n >>> model.learn(total_timesteps=100000)\n \"\"\"\n\n def __init__(self, config=None, agents=None, debug=None):\n super(KoreGymEnv, self).__init__()\n\n if not config:\n config = GAME_CONFIG\n if not agents:\n agents = GAME_AGENTS\n if not debug:\n debug = True\n\n self.agents = agents\n self.env = make(\"kore_fleets\", configuration=config, debug=debug)\n self.config = self.env.configuration\n self.trainer = None\n self.raw_obs = None\n self.previous_obs = None\n\n # Define the action and state space\n # Change these to match your needs. Normalization to the [-1, 1] interval is recommended. See:\n # https://araffin.github.io/slides/rlvs-tips-tricks/#/13/0/0\n # See https://www.gymlibrary.ml/content/spaces/ for more info on OpenAI-gym spaces.\n \n self.action_space = spaces.Discrete(4)\n self.observation_space = spaces.Box(\n low=-1,\n high=1,\n shape=(self.config.size ** 2 * N_FEATURES + 3,),\n dtype=DTYPE\n )\n\n self.strict_reward = config.get('strict', False)\n\n # Debugging info - Enable or disable as needed\n self.reward = 0\n self.n_steps = 0\n self.n_resets = 0\n self.n_dones = 0\n self.last_action = None\n self.last_done = False\n \n self.c_num = 1\n\n def reset(self) -> np.ndarray:\n \"\"\"Resets the trainer and returns the initial observation in state space.\n\n Returns:\n self.obs_as_gym_state: the current observation encoded as a state in state space\n \"\"\"\n # agents = self.agents if np.random.rand() > .5 else self.agents[::-1] # Randomize starting position\n self.trainer = self.env.train(self.agents)\n self.raw_obs = self.trainer.reset()\n self.n_resets += 1\n return self.obs_as_gym_state\n\n def step(self, action: np.ndarray) -> Tuple[np.ndarray, float, bool, Dict]:\n \"\"\"Execute action in the trainer and return the results.\n\n Args:\n action: The action in action space, i.e. the output of the stable-baselines3 agent\n\n Returns:\n self.obs_as_gym_state: the current observation encoded as a state in state space\n reward: The agent's reward\n done: If True, the episode is over\n info: A dictionary with additional debugging information\n \"\"\"\n kore_action = self.gym_to_kore_action(action)\n self.previous_obs = self.raw_obs\n self.raw_obs, _, done, info = self.trainer.step(kore_action) # Ignore trainer reward, which is just delta kore\n self.reward = self.compute_reward(done)\n self.n_steps += 1\n self.last_done = done\n self.last_action = kore_action\n self.n_dones += 1 if done else 0\n\n return self.obs_as_gym_state, self.reward, done, info\n\n def render(self, **kwargs):\n self.env.render(**kwargs)\n\n def close(self):\n pass\n\n @property\n def board(self):\n return Board(self.raw_obs, self.config)\n \n @property\n def board_wrap(self):\n return Board_1st(self.raw_obs, self.config)\n\n @property\n def previous_board(self):\n return Board(self.previous_obs, self.config)\n\n def must_invade(self, obs, config):\n board = Board(obs, config)\n me = board.current_player\n remaining_kore = me.kore\n shipyards = me.shipyards\n convert_cost = board.configuration.convert_cost\n size = board.configuration.size\n spawn_cost = board.configuration.spawn_cost\n turn = board.step\n invading_fleet_size = 50\n invading_fleet_size_buffer = 10\n max_dist = 21\n shipyards = sample(shipyards, len(shipyards))\n for shipyard in shipyards:\n closest_enemy_shipyard = get_closest_enemy_shipyard(board, shipyard.position, board.current_player)\n if closest_enemy_shipyard is None or shipyard.position.distance_to(closest_enemy_shipyard.position, board.configuration.size) > max_dist:\n continue\n if shipyard.ship_count >= invading_fleet_size + invading_fleet_size_buffer:\n flight_plan = get_shortest_flight_path_between(shipyard.position, closest_enemy_shipyard.position, size)\n shipyard.next_action = ShipyardAction.launch_fleet_with_flight_plan(shipyard.ship_count - invading_fleet_size_buffer, flight_plan)\n else:\n shipyard.next_action = spawn_ships(shipyard, remaining_kore, spawn_cost)\n\n def must_invade_concentrate(self, obs, config):\n board = Board(obs, config)\n me = board.current_player\n remaining_kore = me.kore\n shipyards = me.shipyards\n convert_cost = board.configuration.convert_cost\n size = board.configuration.size\n spawn_cost = board.configuration.spawn_cost\n turn = board.step\n invading_fleet_size = 20\n invading_fleet_size_buffer = 0\n max_dist = 21\n enemy_target = None\n closest_dist = max_dist\n for shipyard in shipyards:\n closest_enemy_shipyard = get_closest_enemy_shipyard(board, shipyard.position, board.current_player)\n if closest_enemy_shipyard is not None:\n dist = shipyard.position.distance_to(closest_enemy_shipyard.position, board.configuration.size)\n if dist < closest_dist:\n enemy_target = closest_enemy_shipyard\n closest_dist = dist\n if enemy_target is not None: \n for shipyard in shipyards:\n if shipyard.ship_count >= invading_fleet_size + invading_fleet_size_buffer:\n flight_plan = get_shortest_flight_path_between(shipyard.position, enemy_target.position, size)\n shipyard.next_action = ShipyardAction.launch_fleet_with_flight_plan(shipyard.ship_count - invading_fleet_size_buffer, flight_plan)\n else:\n shipyard.next_action = spawn_ships(shipyard, remaining_kore, spawn_cost)\n else:\n for shipyard in shipyards:\n shipyard.next_action = spawn_ships(shipyard, remaining_kore, spawn_cost)\n \n def must_build(self, obs, config):\n board = Board(obs, config)\n me = board.current_player\n remaining_kore = me.kore\n shipyards = me.shipyards\n convert_cost = board.configuration.convert_cost\n size = board.configuration.size\n spawn_cost = board.configuration.spawn_cost\n turn = board.step\n convert_cost_buffer = 0\n shipyards = sample(shipyards, len(shipyards))\n for shipyard in shipyards:\n if shipyard.ship_count >= convert_cost + convert_cost_buffer and remaining_kore > 500:\n shipyard.next_action = build_new_shipyard(shipyard, board, me, convert_cost)\n \n def gym_to_kore_action(self, gym_action: np.ndarray):\n action = gym_action\n \n obs = self.raw_obs\n if obs[\"step\"] == 0:\n init_logger(logger)\n\n board = self.board_wrap\n step = board.step\n my_id = obs[\"player\"]\n remaining_time = obs[\"remainingOverageTime\"]\n logger.info(f\", remaining_time={remaining_time:.1f}\")\n\n try:\n a = board.get_player(my_id)\n except KeyError:\n return {}\n\n if not a.opponents:\n return {}\n \n if action ==0:\n # offensive\n capture_shipyards(a, max_attack_distance=10)\n adjacent_attack(a, max_distance=10)\n direct_attack(a, max_distance=10)\n defend_shipyards(a)\n expand(a)\n greedy_spawn(a)\n mine(a)\n spawn(a)\n elif action == 1:\n self.must_invade(obs, self.config)\n defend_shipyards(a)\n capture_shipyards(a, max_attack_distance=10)\n adjacent_attack(a, max_distance=10)\n direct_attack(a, max_distance=10)\n expand(a)\n greedy_spawn(a)\n mine(a)\n spawn(a)\n elif action == 2:\n self.must_build(obs, self.config)\n defend_shipyards(a)\n capture_shipyards(a, max_attack_distance=10)\n adjacent_attack(a, max_distance=10)\n direct_attack(a, max_distance=10)\n expand(a)\n greedy_spawn(a)\n mine(a)\n spawn(a)\n elif action == 3:\n self.must_invade_concentrate(obs, self.config)\n defend_shipyards(a)\n capture_shipyards(a, max_attack_distance=10)\n adjacent_attack(a, max_distance=10)\n direct_attack(a, max_distance=10)\n expand(a)\n greedy_spawn(a)\n mine(a)\n spawn(a)\n return a.actions()\n \n @property\n def obs_as_gym_state(self) -> np.ndarray:\n \"\"\"Return the current observation encoded as a state in state space.\n\n In other words, transform a kore observation into a stable-baselines3-compatible np.ndarray.\n\n This property is central - It defines how the kore board is mapped to our state space.\n You can modify it to include as many features as you see convenient.\n\n Let's keep start with something easy: Define a 21x21x4+3 state (size x size x n_features and 3 extra features).\n # Feature 0: How much kore there is in a cell\n # Feature 1: How many ships there are in a cell (>0: friendly, <0: enemy)\n # Feature 2: Fleet direction\n # Feature 3: Is a shipyard present? (1: friendly, -1: enemy, 0: no)\n # Feature 4: Progress - What turn is it?\n # Feature 5: How much kore do I have?\n # Feature 6: How much kore does the opponent have?\n\n We'll make sure that all features are in the range [-1, 1] and as close to a normal distribution as possible.\n\n Note: This mapping doesn't tackle a critical issue in kore: How to encode (full) flight plans?\n \"\"\"\n # Init output state\n gym_state = np.ndarray(shape=(self.config.size, self.config.size, N_FEATURES))\n\n # Get our player ID\n board = self.board\n our_id = board.current_player_id\n\n for point, cell in board.cells.items():\n # Feature 0: How much kore\n gym_state[point.y, point.x, 0] = cell.kore\n\n # Feature 1: How many ships (>0: friendly, <0: enemy)\n # Feature 2: Fleet direction\n fleet = cell.fleet\n if fleet:\n modifier = 1 if fleet.player_id == our_id else -1\n gym_state[point.y, point.x, 1] = modifier * fleet.ship_count\n gym_state[point.y, point.x, 2] = fleet.direction.value\n else:\n # The current cell has no fleet\n gym_state[point.y, point.x, 1] = gym_state[point.y, point.x, 2] = 0\n\n # Feature 3: Shipyard present (1: friendly, -1: enemy)\n shipyard = cell.shipyard\n if shipyard:\n gym_state[point.y, point.x, 3] = 1 if shipyard.player_id == our_id else -1\n else:\n # The current cell has no shipyard\n gym_state[point.y, point.x, 3] = 0\n\n # Normalize features to interval [-1, 1]\n # Feature 0: Logarithmic scale, kore in range [0, MAX_OBSERVABLE_KORE]\n gym_state[:, :, 0] = clip_normalize(\n x=np.log2(gym_state[:, :, 0] + 1),\n low_in=0,\n high_in=np.log2(MAX_OBSERVABLE_KORE)\n )\n\n # Feature 1: Ships in range [-MAX_OBSERVABLE_SHIPS, MAX_OBSERVABLE_SHIPS]\n gym_state[:, :, 1] = clip_normalize(\n x=gym_state[:, :, 1],\n low_in=-MAX_OBSERVABLE_SHIPS,\n high_in=MAX_OBSERVABLE_SHIPS\n )\n\n # Feature 2: Fleet direction in range (1, 4)\n gym_state[:, :, 2] = clip_normalize(\n x=gym_state[:, :, 2],\n low_in=1,\n high_in=4\n )\n\n # Feature 3 is already as normal as it gets\n\n # Flatten the input (recommended by stable_baselines3.common.env_checker.check_env)\n output_state = gym_state.flatten()\n\n # Extra Features: Progress, how much kore do I have, how much kore does opponent have\n player = board.current_player\n opponent = board.opponents[0]\n progress = clip_normalize(board.step, low_in=0, high_in=GAME_CONFIG['episodeSteps'])\n my_kore = clip_normalize(np.log2(player.kore+1), low_in=0, high_in=np.log2(MAX_KORE_IN_RESERVE))\n opponent_kore = clip_normalize(np.log2(opponent.kore+1), low_in=0, high_in=np.log2(MAX_KORE_IN_RESERVE))\n\n return np.append(output_state, [progress, my_kore, opponent_kore])\n\n def compute_reward(self, done: bool, strict=False) -> float:\n \"\"\"Compute the agent reward. Welcome to the fine art of RL.\n\n We'll compute the reward as the current board value and a final bonus if the episode is over. If the player\n wins the episode, we'll add a final bonus that increases with shorter time-to-victory.\n If the player loses, we'll subtract that bonus.\n\n Args:\n done: True if the episode is over\n strict: If True, count only wins/loses (Useful for evaluating a trained agent)\n\n Returns:\n The agent's reward\n \"\"\"\n board = self.board\n previous_board = self.previous_board\n\n if strict:\n if done:\n # Who won?\n # Ugly but 99% sure correct, see https://www.kaggle.com/competitions/kore-2022/discussion/324150#1789804\n agent_reward = self.raw_obs.players[0][0]\n opponent_reward = self.raw_obs.players[1][0]\n return int(agent_reward > opponent_reward)\n else:\n return 0\n else:\n if done:\n # Who won?\n agent_reward = self.raw_obs.players[0][0]\n opponent_reward = self.raw_obs.players[1][0]\n if agent_reward is None or opponent_reward is None:\n we_won = -1\n else:\n we_won = 1 if agent_reward > opponent_reward else -1\n win_reward = we_won * (WIN_REWARD + 100 * (GAME_CONFIG['episodeSteps'] - board.step))\n# win_reward = we_won * WIN_REWARD\n else:\n win_reward = 0\n\n return get_board_value(board) - get_board_value(previous_board) + win_reward\n# return (board.current_player.kore - board.opponents[0].kore) - (previous_board.current_player.kore - previous_board.opponents[0].kore) + win_reward\n\n\ndef clip_normalize(x: Union[np.ndarray, float],\n low_in: float,\n high_in: float,\n low_out=-1.,\n high_out=1.) -> Union[np.ndarray, float]:\n \"\"\"Clip values in x to the interval [low_in, high_in] and then MinMax-normalize to [low_out, high_out].\n\n Args:\n x: The array of float to clip and normalize\n low_in: The lowest possible value in x\n high_in: The highest possible value in x\n low_out: The lowest possible value in the output\n high_out: The highest possible value in the output\n\n Returns:\n The clipped and normalized version of x\n\n Raises:\n AssertionError if the limits are not consistent\n\n Examples:\n >>> clip_normalize(50, low_in=0, high_in=100)\n 0.0\n\n >>> clip_normalize(np.array([-1, .5, 99]), low_in=-1, high_in=1, low_out=0, high_out=2)\n array([0., 1.5, 2.])\n \"\"\"\n assert high_in > low_in and high_out > low_out, \"Wrong limits\"\n\n # Clip outliers\n try:\n x[x > high_in] = high_in\n x[x < low_in] = low_in\n except TypeError:\n x = high_in if x > high_in else x\n x = low_in if x < low_in else x\n\n # y = ax + b\n a = (high_out - low_out) / (high_in - low_in)\n b = high_out - high_in * a\n\n return a * x + b","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:17.007146Z","iopub.execute_input":"2022-07-14T01:43:17.007655Z","iopub.status.idle":"2022-07-14T01:43:17.034287Z","shell.execute_reply.started":"2022-07-14T01:43:17.007620Z","shell.execute_reply":"2022-07-14T01:43:17.032801Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from stable_baselines3 import PPO\nfrom stable_baselines3.common.evaluation import evaluate_policy\nfrom stable_baselines3.common.monitor import Monitor\nfrom environment import KoreGymEnv","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:17.035792Z","iopub.execute_input":"2022-07-14T01:43:17.036766Z","iopub.status.idle":"2022-07-14T01:43:19.935156Z","shell.execute_reply.started":"2022-07-14T01:43:17.036718Z","shell.execute_reply":"2022-07-14T01:43:19.933891Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import random\nimport torch\nimport numpy as np\ndef fix_seed(seed):\n # random\n random.seed(seed)\n # Numpy\n np.random.seed(seed)\n # Pytorch\n torch.manual_seed(seed)\n torch.cuda.manual_seed_all(seed)\n torch.backends.cudnn.deterministic = True\n # Tensorflow\n# tf.random.set_seed(seed)\n# seed = 997269658\nseed = 777\nfix_seed(seed)\n\nlog_dir = \"tmp\"\nkore_env = KoreGymEnv(config=dict(randomSeed=seed)) \nmonitored_env = Monitor(env=kore_env, filename=log_dir)\n\nmodel = PPO(\n 'MlpPolicy', \n monitored_env, \n verbose=1, \n learning_rate=0.0003,\n n_steps=1200, \n batch_size=40,\n n_epochs=10, \n gamma=0.99, \n gae_lambda=0.95, \n clip_range=0.27, \n clip_range_vf=None, \n normalize_advantage=True, \n ent_coef=0.0, \n vf_coef=0.5, \n max_grad_norm=0.5, \n use_sde=False, \n sde_sample_freq=- 1, \n target_kl=None, \n tensorboard_log=None, \n create_eval_env=False, \n policy_kwargs=None, \n seed=None, \n device='auto', \n _init_setup_model=True\n)","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:19.936782Z","iopub.execute_input":"2022-07-14T01:43:19.937855Z","iopub.status.idle":"2022-07-14T01:43:20.079769Z","shell.execute_reply.started":"2022-07-14T01:43:19.937813Z","shell.execute_reply":"2022-07-14T01:43:20.078937Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%time\n# For serious training, likely many more iterations will be needed, as well as hyperparameter tuning!\n# Even so, sometimes training will still fail. RL is like that. Try a couple times with the same config before giving up!\nfrom stable_baselines3.common.callbacks import CheckpointCallback\ncheckpoint_callback = CheckpointCallback(save_freq=400, save_path='./logs/',\n name_prefix='rl_model')\nmodel.learn(total_timesteps=4800, callback=checkpoint_callback) ","metadata":{"execution":{"iopub.status.busy":"2022-07-14T01:43:20.080882Z","iopub.execute_input":"2022-07-14T01:43:20.081390Z","iopub.status.idle":"2022-07-14T02:04:53.699187Z","shell.execute_reply.started":"2022-07-14T01:43:20.081356Z","shell.execute_reply":"2022-07-14T02:04:53.698138Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Watch it mercilessly beat the baseline bot - Note: The current episode might not be over yet\nkore_env.render(mode=\"ipython\", width=1000, height=800)","metadata":{"execution":{"iopub.status.busy":"2022-07-14T02:04:53.700793Z","iopub.execute_input":"2022-07-14T02:04:53.701451Z","iopub.status.idle":"2022-07-14T02:04:54.223570Z","shell.execute_reply.started":"2022-07-14T02:04:53.701405Z","shell.execute_reply":"2022-07-14T02:04:54.222729Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model.save(\"baseline_agent\")","metadata":{"execution":{"iopub.status.busy":"2022-07-14T02:04:54.224784Z","iopub.execute_input":"2022-07-14T02:04:54.225265Z","iopub.status.idle":"2022-07-14T02:04:54.252061Z","shell.execute_reply.started":"2022-07-14T02:04:54.225232Z","shell.execute_reply":"2022-07-14T02:04:54.250855Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import numpy as np\n\neval_env = KoreGymEnv(config=dict(strict=True)) # The 'strict' flags sets rewards to 1 if the agent won the episode and 0 else. Useful for evaluation.\nmonitored_env = Monitor(env=eval_env)\nmodel_loaded = PPO.load('baseline_agent')\n\ndef evaluate(model, num_episodes=1):\n \"\"\"\n Evaluate a RL agent - Adapted from \n https://colab.research.google.com/github/Stable-Baselines-Team/rl-colab-notebooks/blob/sb3/stable_baselines_getting_started.ipynb\n :param model: (BaseRLModel object) the RL Agent\n :param num_episodes: (int) number of episodes to evaluate it\n :return: (float) Mean reward for the last num_episodes\n \"\"\"\n all_episode_rewards = []\n for i in range(num_episodes):\n episode_rewards = []\n done = False\n obs = monitored_env.reset()\n while not done:\n action, _ = model.predict(obs)\n obs, _, done, info = monitored_env.step(action)\n if done:\n agent_reward = monitored_env.env.raw_obs.players[0][0]\n opponent_reward = monitored_env.env.raw_obs.players[1][0]\n reward = agent_reward > opponent_reward\n else:\n reward = 0\n # print(reward)\n # monitored_env.render(mode='ipython', height=400, width=300)\n episode_rewards.append(reward)\n\n all_episode_rewards.append(sum(episode_rewards))\n\n mean_episode_reward = np.mean(all_episode_rewards)\n print(\"Mean reward:\", mean_episode_reward, \"Num episodes:\", num_episodes)\n\n return mean_episode_reward\n\n# evaluate(model_loaded, 20)","metadata":{"execution":{"iopub.status.busy":"2022-07-14T02:04:54.255613Z","iopub.execute_input":"2022-07-14T02:04:54.256056Z","iopub.status.idle":"2022-07-14T02:04:54.337830Z","shell.execute_reply.started":"2022-07-14T02:04:54.256003Z","shell.execute_reply":"2022-07-14T02:04:54.336750Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%writefile main.py\n# All this syspath wranglig is needed to make sure that the agent runs on the target environment and can load both the external dependencies\n# and the saved model. Dear kaggle, if possible, please make this easier!\nimport os\nimport sys\nKAGGLE_AGENT_PATH = \"/kaggle_simulations/agent/\"\nif os.path.exists(KAGGLE_AGENT_PATH):\n # We're in the kaggle target system\n sys.path.insert(0, os.path.join(KAGGLE_AGENT_PATH, 'lib'))\n agent_path = os.path.join(KAGGLE_AGENT_PATH, 'baseline_agent')\nelse:\n # We're somewhere else\n sys.path.insert(0, os.path.join(os.getcwd(), 'lib'))\n agent_path = 'baseline_agent'\n\n# Now for the actual agent\nfrom stable_baselines3 import PPO\nfrom environment import KoreGymEnv\n\nmodel = PPO.load(agent_path)\nkore_env = KoreGymEnv()\n\ndef agent(obs, config):\n kore_env.raw_obs = obs\n state = kore_env.obs_as_gym_state\n action, _ = model.predict(state)\n return kore_env.gym_to_kore_action(action)","metadata":{"execution":{"iopub.status.busy":"2022-07-14T02:04:54.339198Z","iopub.execute_input":"2022-07-14T02:04:54.339568Z","iopub.status.idle":"2022-07-14T02:04:54.345990Z","shell.execute_reply.started":"2022-07-14T02:04:54.339535Z","shell.execute_reply":"2022-07-14T02:04:54.345200Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"!tar -czf submission.tar.gz main.py config.py environment.py reward_utils.py extra_helpers.py defend.py attack.py build.py mine.py basic_1st.py board_1st.py logger_1st.py offence_1st.py defence_1st.py expantion_1st.py geometry_1st.py mining_1st.py control_1st.py helpers_1st.py baseline_agent.zip lib","metadata":{"execution":{"iopub.status.busy":"2022-07-14T02:04:54.347665Z","iopub.execute_input":"2022-07-14T02:04:54.348315Z","iopub.status.idle":"2022-07-14T02:04:55.510589Z","shell.execute_reply.started":"2022-07-14T02:04:54.348267Z","shell.execute_reply":"2022-07-14T02:04:55.509472Z"},"trusted":true},"execution_count":null,"outputs":[]}]}