Add lux

new_src/.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+submission.tar.gz

new_src/agent.py

@@ -0,0 +1,104 @@
+from lux.kit import obs_to_game_state, GameState, EnvConfig
+from lux.utils import direction_to, my_turn_to_place_factory
+import numpy as np
+import sys
+class Agent():
+    def __init__(self, player: str, env_cfg: EnvConfig) -> None:
+        self.player = player
+        self.opp_player = "player_1" if self.player == "player_0" else "player_0"
+        np.random.seed(0)
+        self.env_cfg: EnvConfig = env_cfg
+        self.factory_score = None
+
+    def early_setup(self, step: int, obs, remainingOverageTime: int = 60):
+        if step == 0:
+            # bid 0 to not waste resources bidding and declare as the default faction
+            # you can bid -n to prefer going second or n to prefer going first in placement
+            return dict(faction="AlphaStrike", bid=0)
+        else:
+            game_state = obs_to_game_state(step, self.env_cfg, obs)
+            # factory placement period
+            if self.factory_score is None:
+                map_size = self.env_cfg.map_size
+                self.factory_score = np.zeros((map_size, map_size))
+                self.factory_score += conv2d(game_state.board.rubble, average_kernel(5), n=3)
+                ice_tile_locations = np.argwhere(game_state.board.ice == 1)
+                all_locations = np.mgrid[:map_size, :map_size].swapaxes(0, 2).reshape(-1, 2)
+                distances = np.linalg.norm(np.expand_dims(all_locations, 1) - np.expand_dims(ice_tile_locations, 0), ord=1, axis=-1)
+                distances = np.min(distances, axis=-1)
+                distances = distances.reshape(map_size, map_size)
+                # distances[x][y] is the distance to the nearest ice tile 
+                self.factory_score += distances
+
+            # how much water and metal you have in your starting pool to give to new factories
+            water_left = game_state.teams[self.player].water
+            metal_left = game_state.teams[self.player].metal
+
+            # how many factories you have left to place
+            factories_to_place = game_state.teams[self.player].factories_to_place
+            # whether it is your turn to place a factory
+            my_turn_to_place = my_turn_to_place_factory(game_state.teams[self.player].place_first, step)
+            if factories_to_place > 0 and my_turn_to_place:
+                # we will spawn our factory in a random location with 150 metal and water if it is our turn to place
+                factory_score = self.factory_score + (obs["board"]["valid_spawns_mask"] == 0) * 1e9
+                spawn_loc = np.argmin(factory_score)
+                map_size = self.env_cfg.map_size
+                spawn_loc = np.array([spawn_loc // map_size, spawn_loc % map_size])
+                return dict(spawn=spawn_loc, metal=150, water=150)
+            return dict()
+
+    def act(self, step: int, obs, remainingOverageTime: int = 60):
+        actions = dict()
+        game_state = obs_to_game_state(step, self.env_cfg, obs)
+        factories = game_state.factories[self.player]
+        factory_tiles, factory_units = [], []
+        for unit_id, factory in factories.items():
+            if factory.power >= self.env_cfg.ROBOTS["HEAVY"].POWER_COST and \
+            factory.cargo.metal >= self.env_cfg.ROBOTS["HEAVY"].METAL_COST:
+                actions[unit_id] = factory.build_heavy()
+            factory_tiles += [factory.pos]
+            factory_units += [factory]
+        factory_tiles = np.array(factory_tiles)
+
+        units = game_state.units[self.player]
+        ice_map = game_state.board.ice
+        ice_tile_locations = np.argwhere(ice_map == 1)
+        for unit_id, unit in units.items():
+
+            # track the closest factory
+            closest_factory = None
+            adjacent_to_factory = False
+            if len(factory_tiles) > 0:
+                factory_distances = np.linalg.norm(factory_tiles - unit.pos, ord=1, axis=1)
+                closest_factory_tile = factory_tiles[np.argmin(factory_distances)]
+                closest_factory = factory_units[np.argmin(factory_distances)]
+                adjacent_to_factory = np.linalg.norm(closest_factory_tile - unit.pos, ord=1) == 0
+
+                ice_threshold = 40
+                # previous ice mining code
+                if adjacent_to_factory and unit.power < unit.unit_cfg.INIT_POWER:
+                    actions[unit_id] = [unit.pickup(4, unit.unit_cfg.BATTERY_CAPACITY, repeat=0, n=1)]
+                    # 4 means power
+                elif unit.cargo.ice < ice_threshold:
+                    ice_tile_distances = np.linalg.norm(ice_tile_locations - unit.pos, ord=1, axis=1)
+                    closest_ice_tile = ice_tile_locations[np.argmin(ice_tile_distances)]
+                    if np.all(closest_ice_tile == unit.pos):
+                        if unit.power >= unit.dig_cost(game_state) + unit.action_queue_cost(game_state):
+                            actions[unit_id] = [unit.dig(repeat=0, n=1)]
+                    else:
+                        direction = direction_to(unit.pos, closest_ice_tile)
+                        move_cost = unit.move_cost(game_state, direction)
+                        if move_cost is not None and unit.power >= move_cost + unit.action_queue_cost(game_state):
+                            actions[unit_id] = [unit.move(direction, repeat=0, n=1)]
+                # else if we have enough ice, we go back to the factory and dump it.
+                elif unit.cargo.ice >= ice_threshold:
+                    direction = direction_to(unit.pos, closest_factory_tile)
+                    if adjacent_to_factory:
+                        if unit.power >= unit.action_queue_cost(game_state):
+                            actions[unit_id] = [unit.transfer(direction, 0, unit.cargo.ice, repeat=0, n=1)]
+
+                    else:
+                        move_cost = unit.move_cost(game_state, direction)
+                        if move_cost is not None and unit.power >= move_cost + unit.action_queue_cost(game_state):
+                            actions[unit_id] = [unit.move(direction, repeat=0, n=1)]
+        return actions

new_src/lux/cargo.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+
+@dataclass
+class UnitCargo:
+    ice: int = 0
+    ore: int = 0
+    water: int = 0
+    metal: int = 0

new_src/lux/config.py

@@ -0,0 +1,136 @@
+import dataclasses
+from argparse import Namespace
+from dataclasses import dataclass
+from typing import Dict, List
+
+
+def convert_dict_to_ns(x):
+    if isinstance(x, dict):
+        for k in x:
+            x[k] = convert_dict_to_ns(x)
+        return Namespace(x)
+
+
+@dataclass
+class UnitConfig:
+    METAL_COST: int = 100
+    POWER_COST: int = 500
+    CARGO_SPACE: int = 1000
+    BATTERY_CAPACITY: int = 1500
+    CHARGE: int = 1
+    INIT_POWER: int = 50
+    MOVE_COST: int = 1
+    RUBBLE_MOVEMENT_COST: float = 1
+    DIG_COST: int = 5
+    DIG_RUBBLE_REMOVED: int = 1
+    DIG_RESOURCE_GAIN: int = 2
+    DIG_LICHEN_REMOVED: int = 10
+    SELF_DESTRUCT_COST: int = 10
+    RUBBLE_AFTER_DESTRUCTION: int = 1
+    ACTION_QUEUE_POWER_COST: int = 1
+
+
+@dataclass
+class EnvConfig:
+    ## various options that can be configured if needed
+
+    ### Variable parameters that don't affect game logic much ###
+    max_episode_length: int = 1000
+    map_size: int = 64
+    verbose: int = 1
+
+    # this can be disabled to improve env FPS but assume your actions are well formatted
+    # During online competition this is set to True
+    validate_action_space: bool = True
+
+    ### Constants ###
+    # you can only ever transfer in/out 1000 as this is the max cargo space.
+    max_transfer_amount: int = 10000
+    MIN_FACTORIES: int = 4
+    MAX_FACTORIES: int = 10
+    CYCLE_LENGTH: int = 50
+    DAY_LENGTH: int = 30
+    UNIT_ACTION_QUEUE_SIZE: int = 20  # when set to 1, then no action queue is used
+
+    MAX_RUBBLE: int = 100
+    FACTORY_RUBBLE_AFTER_DESTRUCTION: int = 50
+    INIT_WATER_METAL_PER_FACTORY: int = (
+        150  # amount of water and metal units given to each factory
+    )
+    INIT_POWER_PER_FACTORY: int = 1000
+
+    #### LICHEN ####
+    MIN_LICHEN_TO_SPREAD: int = 20
+    LICHEN_LOST_WITHOUT_WATER: int = 1
+    LICHEN_GAINED_WITH_WATER: int = 1
+    MAX_LICHEN_PER_TILE: int = 100
+    POWER_PER_CONNECTED_LICHEN_TILE: int = 1
+
+    # cost of watering with a factory is `ceil(# of connected lichen tiles) / (this factor) + 1`
+    LICHEN_WATERING_COST_FACTOR: int = 10
+
+    #### Bidding System ####
+    BIDDING_SYSTEM: bool = True
+
+    #### Factories ####
+    FACTORY_PROCESSING_RATE_WATER: int = 100
+    ICE_WATER_RATIO: int = 4
+    FACTORY_PROCESSING_RATE_METAL: int = 50
+    ORE_METAL_RATIO: int = 5
+    # game design note: Factories close to resource cluster = more resources are refined per turn
+    # Then the high ice:water and ore:metal ratios encourages transfer of refined resources between
+    # factories dedicated to mining particular clusters which is more possible as it is more compact
+
+    FACTORY_CHARGE: int = 50
+    FACTORY_WATER_CONSUMPTION: int = 1
+    # game design note: with a positve water consumption, game becomes quite hard for new competitors.
+    # so we set it to 0
+
+    #### Collision Mechanics ####
+    POWER_LOSS_FACTOR: float = 0.5
+
+    #### Units ####
+    ROBOTS: Dict[str, UnitConfig] = dataclasses.field(
+        default_factory=lambda: dict(
+            LIGHT=UnitConfig(
+                METAL_COST=10,
+                POWER_COST=50,
+                INIT_POWER=50,
+                CARGO_SPACE=100,
+                BATTERY_CAPACITY=150,
+                CHARGE=1,
+                MOVE_COST=1,
+                RUBBLE_MOVEMENT_COST=0.05,
+                DIG_COST=5,
+                SELF_DESTRUCT_COST=5,
+                DIG_RUBBLE_REMOVED=2,
+                DIG_RESOURCE_GAIN=2,
+                DIG_LICHEN_REMOVED=10,
+                RUBBLE_AFTER_DESTRUCTION=1,
+                ACTION_QUEUE_POWER_COST=1,
+            ),
+            HEAVY=UnitConfig(
+                METAL_COST=100,
+                POWER_COST=500,
+                INIT_POWER=500,
+                CARGO_SPACE=1000,
+                BATTERY_CAPACITY=3000,
+                CHARGE=10,
+                MOVE_COST=20,
+                RUBBLE_MOVEMENT_COST=1,
+                DIG_COST=60,
+                SELF_DESTRUCT_COST=100,
+                DIG_RUBBLE_REMOVED=20,
+                DIG_RESOURCE_GAIN=20,
+                DIG_LICHEN_REMOVED=100,
+                RUBBLE_AFTER_DESTRUCTION=10,
+                ACTION_QUEUE_POWER_COST=10,
+            ),
+        )
+    )
+
+    @classmethod
+    def from_dict(cls, data):
+        data["ROBOTS"]["LIGHT"] = UnitConfig(**data["ROBOTS"]["LIGHT"])
+        data["ROBOTS"]["HEAVY"] = UnitConfig(**data["ROBOTS"]["HEAVY"])
+        return cls(**data)

new_src/lux/factory.py

@@ -0,0 +1,55 @@
+import math
+from sys import stderr
+import numpy as np
+from dataclasses import dataclass
+from lux.cargo import UnitCargo
+from lux.config import EnvConfig
+@dataclass
+class Factory:
+    team_id: int
+    unit_id: str
+    strain_id: int
+    power: int
+    cargo: UnitCargo
+    pos: np.ndarray
+    # lichen tiles connected to this factory
+    # lichen_tiles: np.ndarray
+    env_cfg: EnvConfig
+
+    def build_heavy_metal_cost(self, game_state):
+        unit_cfg = self.env_cfg.ROBOTS["HEAVY"]
+        return unit_cfg.METAL_COST
+    def build_heavy_power_cost(self, game_state):
+        unit_cfg = self.env_cfg.ROBOTS["HEAVY"]
+        return unit_cfg.POWER_COST
+    def can_build_heavy(self, game_state):
+        return self.power >= self.build_heavy_power_cost(game_state) and self.cargo.metal >= self.build_heavy_metal_cost(game_state)
+    def build_heavy(self):
+        return 1
+
+    def build_light_metal_cost(self, game_state):
+        unit_cfg = self.env_cfg.ROBOTS["LIGHT"]
+        return unit_cfg.METAL_COST
+    def build_light_power_cost(self, game_state):
+        unit_cfg = self.env_cfg.ROBOTS["LIGHT"]
+        return unit_cfg.POWER_COST
+    def can_build_light(self, game_state):
+        return self.power >= self.build_light_power_cost(game_state) and self.cargo.metal >= self.build_light_metal_cost(game_state)
+
+    def build_light(self):
+        return 0
+
+    def water_cost(self, game_state):
+        """
+        Water required to perform water action
+        """
+        owned_lichen_tiles = (game_state.board.lichen_strains == self.strain_id).sum()
+        return np.ceil(owned_lichen_tiles / self.env_cfg.LICHEN_WATERING_COST_FACTOR)
+    def can_water(self, game_state):
+        return self.cargo.water >= self.water_cost(game_state)
+    def water(self):
+        return 2
+
+    @property
+    def pos_slice(self):
+        return slice(self.pos[0] - 1, self.pos[0] + 2), slice(self.pos[1] - 1, self.pos[1] + 2)

new_src/lux/forward_sim.py

@@ -0,0 +1,27 @@
+def forward_sim(full_obs, env_cfg, n=2):
+    """
+    Forward sims for `n` steps given the current full observation and env_cfg
+
+    If forward sim leads to the end of a game, it won't return any additional observations, just the original one
+    """
+    from luxai_s2 import LuxAI_S2
+    from luxai_s2.config import UnitConfig
+    import copy
+    agent = "player_0"
+    env = LuxAI_S2(collect_stats=False, verbose=0)
+    env.reset(seed=0)
+    env.state = env.state.from_obs(full_obs, env_cfg)
+    env.env_cfg = env.state.env_cfg
+    env.env_cfg.verbose = 0
+    env.env_steps = env.state.env_steps
+    forward_obs = [full_obs]
+    for _ in range(n):
+        empty_actions = dict()
+        for agent in env.agents:
+            empty_actions[agent] = dict()
+        if len(env.agents) == 0:
+            # can't step any further
+            return [full_obs]
+        obs, _, _, _, _ = env.step(empty_actions)
+        forward_obs.append(obs[agent])
+    return forward_obs