diff --git a/__pycache__/coreHelper.cpython-311.pyc b/__pycache__/coreHelper.cpython-311.pyc new file mode 100644 index 0000000..c29a9e8 Binary files /dev/null and b/__pycache__/coreHelper.cpython-311.pyc differ diff --git a/__pycache__/defend.cpython-311.pyc b/__pycache__/defend.cpython-311.pyc new file mode 100644 index 0000000..7e8bb18 Binary files /dev/null and b/__pycache__/defend.cpython-311.pyc differ diff --git a/__pycache__/homedefense.cpython-311.pyc b/__pycache__/homedefense.cpython-311.pyc new file mode 100644 index 0000000..32132c5 Binary files /dev/null and b/__pycache__/homedefense.cpython-311.pyc differ diff --git a/__pycache__/main.cpython-311.pyc b/__pycache__/main.cpython-311.pyc new file mode 100644 index 0000000..0e8de04 Binary files /dev/null and b/__pycache__/main.cpython-311.pyc differ diff --git a/__pycache__/mapanalysis.cpython-311.pyc b/__pycache__/mapanalysis.cpython-311.pyc new file mode 100644 index 0000000..d2d8890 Binary files /dev/null and b/__pycache__/mapanalysis.cpython-311.pyc differ diff --git a/__pycache__/turretplan.cpython-311.pyc b/__pycache__/turretplan.cpython-311.pyc new file mode 100644 index 0000000..4086e01 Binary files /dev/null and b/__pycache__/turretplan.cpython-311.pyc differ diff --git a/bot.zip b/bot.zip index 502ca8f..9d1853f 100644 Binary files a/bot.zip and b/bot.zip differ diff --git a/coreHelper.py b/coreHelper.py deleted file mode 100644 index b095a30..0000000 --- a/coreHelper.py +++ /dev/null @@ -1,32 +0,0 @@ -from fcode import * - - -def spawnBots(numSpawned: int, ct: Controller) -> bool: - currentTitanium = ct.get_global_resources() - if numSpawned < 5 and currentTitanium >= 350: - return True - if numSpawned > 4 and currentTitanium >= 400 * ct.get_scale_percent() / 100: - return True - return False - - -def findThreats(ct: Controller) -> list: - threats = [] - curRound = ct.get_current_round() - for i in ct.get_nearby_buildings(8): - if (ct.get_entity_type(i) == EntityType.LAUNCHER and - ct.get_team(i) != ct.get_team()): - threats.append(ct.get_position(i)) - return threats - - -def findTeamCore(ct: Controller): - for i in ct.get_nearby_buildings(): - if ct.get_entity_type(i) == EntityType.CORE and ct.get_team(i) == ct.get_team(): - return ct.get_position(i) - - -def findEnemyCore(ct: Controller): - for i in ct.get_nearby_buildings(): - if ct.get_entity_type(i) == EntityType.CORE and ct.get_team(i) != ct.get_team(): - return ct.get_position(i) diff --git a/defend.py b/defend.py deleted file mode 100644 index 0df8efc..0000000 --- a/defend.py +++ /dev/null @@ -1,119 +0,0 @@ -from fcode import * -import random - - -class Defender: - def __init__(self): - self.defendPos = None - self.destroyTurns = 0 - - def defendInfrastructure(self, ct: Controller, teamCore, pf): - myLoc = ct.get_position() - if self.defendPos is None: - for i in ct.get_nearby_buildings(): - if ct.get_team(i) == ct.get_team() and ct.get_hp(i) < ct.get_max_hp(i): - self.defendPos = ct.get_position(i) - break - if self.defendPos is not None: - if myLoc.distance_squared(self.defendPos) > 2: - pf.moveTo(ct, self.defendPos) - self.placeLauncher(ct, self.defendPos) - if ct.can_heal(self.defendPos): - ct.heal(self.defendPos) - - if self.defendPos is not None: - if ct.is_in_vision(self.defendPos): - defendID = ct.get_tile_building_id(self.defendPos) - if defendID is None: - self.defendPos = None - elif ct.get_hp(defendID) == ct.get_max_hp(defendID): - self.defendPos = None - else: - self.defendPos = None - - def placeLauncher(self, ct: Controller, pos: Position): - myLoc = ct.get_position() - for i in ct.get_nearby_buildings(): - if ct.get_entity_type(i) == EntityType.LAUNCHER and ct.get_team(i) == ct.get_team(): - if ct.get_position(i).distance_squared(pos) <= 2: - return # Launcher already built - topCands = [] - otherCands = [] - for k in ct.get_nearby_tiles(2): - if k.distance_squared(pos) <= 2 and k.distance_squared(myLoc) <= 2: - bid = ct.get_tile_building_id(k) - if bid is None and ct.get_tile_env(k) != Environment.WALL: - topCands.append(k) - elif bid is not None and ct.get_team(bid) == ct.get_team(): - etype_k = ct.get_entity_type(bid) - if etype_k not in (EntityType.CONVEYOR, EntityType.HARVESTER, EntityType.SENTINEL): - otherCands.append(k) - for a in topCands: - if ct.can_build_launcher(a): - if self.destroyTurns > 5: - self.destroyTurns = 0 - ct.build_launcher(a) - return - didDestroy = False - for a in otherCands: - if ct.can_destroy(a) and didDestroy == False: - ct.destroy(a) - didDestroy = True - if ct.can_build_launcher(a): - if self.destroyTurns > 5: - self.destroyTurns = 0 - ct.build_launcher(a) - return - return - - def picHealingLoc(self, ct: Controller, teamCore: Position, alwaysDefense: bool = False) -> Position | None: - candidates = [] - for i in ct.get_nearby_tiles(16): - if ct.get_tile_env(i) == Environment.EMPTY and i.distance_squared(teamCore) < ((ct.get_map_width() + ct.get_map_height() - 8) // 4): - bid = ct.get_tile_building_id(i) - if bid is None: - candidates.append(i) - else: - entype = ct.get_entity_type(bid) - if (entype == EntityType.CONVEYOR or entype == EntityType.CORE): - candidates.append(i) - - if not candidates: - return Position(teamCore.x + random.randint(-1, 2), teamCore.y + random.randint(-1, 2)) - - buildingweights = [] - buildingpos = [] - for bid in ct.get_nearby_buildings(): - if ct.get_team(bid) == ct.get_team(): - entype = ct.get_entity_type(bid) - if entype == EntityType.CONVEYOR: - buildingweights.append(25) - buildingpos.append(ct.get_position(bid)) - elif entype == EntityType.LAUNCHER: - buildingweights.append(5) - buildingpos.append(ct.get_position(bid)) - elif entype == EntityType.CORE: - buildingweights.append(100) - buildingpos.append(ct.get_position(bid)) - - bestTile = None - bestScore = -1 - - for i in candidates: - score = 0 - for j in range(len(buildingweights)): - pose = buildingpos[j] - weight = buildingweights[j] - dist = i.distance_squared(pose) - if dist < 50: - score += max(weight * (1 - 0.02 * dist), 0) - - if score > bestScore: - bestTile = i - bestScore = score - if alwaysDefense == True: - return Position(teamCore.x + random.randint(-1, 1), teamCore.y + random.randint(-1, 1)) - else: - if bestTile is None: - return Position(teamCore.x + random.randint(-1, 1), teamCore.y + random.randint(-1, 1)) - return Position(bestTile.x + random.randint(-1, 1), bestTile.y + random.randint(-1, 1)) diff --git a/main.py b/main.py index abee683..5fdbc93 100644 --- a/main.py +++ b/main.py @@ -1,736 +1,1443 @@ -from fcode import * -from coreHelper import * -from pathfinding import * -from defend import * -import math -import random -# This file contains the eco bot code, ported from Cambridge Battlecode to -# the Florent Code League (fcode) API. +"""Khaos rebuild — memoryless state-scoring bot (Pantheon architecture). -Directions = [d for d in Direction if d != Direction.CENTRE] -DIRECTIONS = Directions # alias used in attack() -CardDirections = [Direction.NORTH, Direction.EAST, Direction.SOUTH, Direction.WEST] -# 20 ammo = 10 gunner shots or 2 sentinel shots. -GLOBAL_AMMO_BUFFER = 20 +Every builder, every turn: score seven candidate states in descending ceiling +order with early exit, run the single best. No mode flags, no multi-turn +commitments — targets are re-derived each turn from the bitmask map layer, +deconflicted through a Voronoi claim partition, with TTL caches for targets +that recently failed. +States (ceiling): attack 9 · heal 8 · route 7.75 · secure 7.5 · harvest 4 · +disrupt 2 · explore 1. +""" -class Player: - def __init__(self): - # Other Files - self.pf = Pathfinder() - self.defender = Defender() +import random - self.numSpawned = 0 +from fcode import Controller, Direction, EntityType, Environment, Position - # Role - self.attackMode = False - self.alwaysDefense = False - self.defenseMode = False - self.justSpawned = True +import mapanalysis +import mapinfo as mapinfo_mod +import path +import planes +import symmetry - # Attacking - self.attackPos = None - self.spawnSentinelHere = None +CARDINALS = [Direction.NORTH, Direction.SOUTH, Direction.EAST, Direction.WEST] +DELTA_TO_DIR = { + (0, -1): Direction.NORTH, (0, 1): Direction.SOUTH, + (1, 0): Direction.EAST, (-1, 0): Direction.WEST, +} + +GLOBAL_AMMO_BUFFER = 20 +MAX_OPEN_BUILDERS = 5 +TTL_ROUNDS = 100 +CPU_SOFT_STOP = 6500 # stop scoring lower states past this many us - # Cores - self.teamCore = None # reference corner tile of the (2x2) core - self.teamCoreTiles = None # all tiles of the core footprint - self.enemyCore = None - # Launcher Proofing - self.oldPos = None - - # Harvester Finding - self.currentHarvester = None - self.foundHarvester = None - self.usedHarvesters = [] - self.reachHarvesterTime = 0 - - # Building Back + Sentinel Placment - self.conveyorEnd = None - self.surroundHarvester = False - self.protectTurns = 0 - self.sentinelSpot = None - self.isEnemyInfastructure = None - self.nextTurretCountDown = 0 - - # Placing Gunner On Attack - self.turretTimeOut = 0 # Starts when turret is placed - self.enemyPos = None - self.enemyType = -1 - # Exploring - self.explorePos = None - self.exploreTime = 0 - self.attackMoveTimeout = 0 - self.attackHypothesisIndex = 0 # opp core symmetry change +def manhattan(a, b): + return abs(a.x - b.x) + abs(a.y - b.y) + +class Player: + def __init__(self): + self.mi = None + self.myNum = -1 + self.numSpawned = 0 + self.teamCore = None + self.teamCoreTiles = None + self.enemyCore = None + self.symTracker = None + self.fullmap = None + self.fullmapVer = -1 + self.routeField = None + self.routeVer = -1 + self._connected = set() # own conveyors whose chain reaches the core + self._homeDisc = None # cached home-guard disc for builder #1 + self.brokenBarriers = [] # own barriers we broke while walking + self._plantCache = None # score-plane best placement cache + self.enemyField = None # dist-to-enemy-core field (attack route) + self.enemyFieldVer = -1 + self.ttl = {} # (state, x, y) -> expiry round + self.stuck = 0 + self.lastPos = None + self.exploreTarget = None + # turret / launcher state + self.idleTurns = 0 + self.enemyBotHistory = {} + self.tapWait = 0 # sentinel one-tap coordination hold counter + + # ------------------------------------------------------------------ def run(self, ct: Controller) -> None: etype = ct.get_entity_type() if etype == EntityType.CORE: self.core(ct) elif etype == EntityType.BUILDER_BOT: - self.builderBot(ct) + self.builder(ct) elif etype == EntityType.SENTINEL: self.sentinel(ct) - elif etype == EntityType.LAUNCHER: - self.launcher(ct) elif etype == EntityType.GUNNER: self.gunner(ct) + elif etype == EntityType.LAUNCHER: + self.launcher(ct) + # ------------------------------------------------------------------ + # Core: fan-spawn opening, defensive spawn, ammo buffer + # ------------------------------------------------------------------ def core(self, ct: Controller) -> None: - # spawn first so ammo conversion cannot prevent this turns spawn - if spawnBots(self.numSpawned, ct): - corePos = ct.get_position() - threats = findThreats(ct) - spawned = False - + import math + pos = ct.get_position() + if self.mi is None: + w, h = ct.get_map_width(), ct.get_map_height() + self.mi = mapinfo_mod.MapInfo(w, h) + mapanalysis.reset_map(w, h) + rnd = ct.get_current_round() + ti = ct.get_global_resources() + # Pantheon spawn policy: baseline 200 until 12 allies, then 400 — + # spawn whenever titanium clears baseline + the builder's own cost. + # (Our old 350/400*scale gates stalled workforce growth exactly + # during establishment.) + baseline = 400 if ct.get_unit_count() >= 12 else 200 + wantSpawn = ( + ti >= baseline + ct.get_builder_bot_cost() + or (rnd >= 30 and ct.get_unit_count() <= 6 + and ti >= ct.get_builder_bot_cost() + 60)) + if wantSpawn: + candidates = [] for dx in (-1, 0, 1, 2): for dy in (-1, 0, 1, 2): - # skip the four tiles occupied by the core if 0 <= dx <= 1 and 0 <= dy <= 1: continue - - spawnPos = Position( - corePos.x + dx, - corePos.y + dy, - ) - - if not ct.can_spawn(spawnPos): - continue - - isThreat = any( - spawnPos.distance_squared(threat) <= 2 - for threat in threats - ) - if isThreat: + p = Position(pos.x + dx, pos.y + dy) + if ct.can_spawn(p): + candidates.append(p) + if candidates: + ccx, ccy = pos.x + 0.5, pos.y + 0.5 + pick = None + # defensive: intruder at the base with no ally on it + enemyPos, enemyDist, allies = None, 1e9, [] + for eid in ct.get_nearby_units(): + ep = ct.get_position(eid) + if ct.get_team(eid) == ct.get_team(): + if ct.get_entity_type(eid) == EntityType.BUILDER_BOT: + allies.append(ep) continue - - ct.spawn_builder(spawnPos) - self.numSpawned += 1 - spawned = True - break - - if spawned: + d = (ep.x - ccx) ** 2 + (ep.y - ccy) ** 2 + if d < enemyDist: + enemyDist, enemyPos = d, ep + if enemyPos is not None and enemyDist <= 18 and not any( + a.distance_squared(enemyPos) <= 8 for a in allies): + pick = min(candidates, key=lambda p: manhattan(p, enemyPos)) + elif self.numSpawned < 4: + w, h = ct.get_map_width(), ct.get_map_height() + primary = math.atan2(h / 2 - ccy, w / 2 - ccx) + target = primary + (-1.178, -0.393, 0.393, 1.178)[self.numSpawned] + tx, ty = math.cos(target), math.sin(target) + + def align(p): + vx, vy = p.x + 0.5 - ccx, p.y + 0.5 - ccy + norm = math.hypot(vx, vy) or 1.0 + return -(vx * tx + vy * ty) / norm + pick = min(candidates, key=align) + else: + pick = candidates[0] + ct.spawn_builder(pick) + self.numSpawned += 1 + ct.write_store(0, self.numSpawned) + # distress: hurt core or enemy building in core vision -> broadcast on + # slot 7 so fanned-out builders converge to heal/counter-battery. + distress = ct.get_hp() < ct.get_max_hp() + if not distress: + for bid in ct.get_nearby_buildings(): + if ct.get_team(bid) != ct.get_team(): + distress = True break - - # Turrets from ammo pool instead - currentAmmo = ct.get_global_ammo() - - if currentAmmo < GLOBAL_AMMO_BUFFER: - # preserve titn to afford another builder - spareTitanium = ( - ct.get_global_resources() - - ct.get_builder_bot_cost() - ) - - amount = min( - GLOBAL_AMMO_BUFFER - currentAmmo, - spareTitanium, - ) - + if distress: + ct.write_store(7, rnd) + # ammo buffer (raised while distressed so counter-turrets can fire) + buffer = 40 if distress else GLOBAL_AMMO_BUFFER + cur = ct.get_global_ammo() + if cur < buffer: + spare = ct.get_global_resources() - ct.get_builder_bot_cost() + amount = min(buffer - cur, spare) if amount > 0 and ct.can_convert_ammo(amount): ct.convert_ammo(amount) - # new for ts - def gunner(self, ct: Controller) -> None: - target = ct.get_gunner_target() - - if target is not None and ct.can_fire(target): - ct.fire(target) - - def sentinel(self, ct: Controller) -> None: + # ------------------------------------------------------------------ + # Builder: the memoryless spine + # ------------------------------------------------------------------ + def builder(self, ct: Controller) -> None: + mi = self._setup(ct) + mi.update_vision(ct) + mi.absorb(ct) + self._updateCores(ct) + self._updateSymmetry(ct) + + rnd = ct.get_current_round() + pos = ct.get_position() + if self.lastPos == pos: + self.stuck += 1 + else: + self.stuck = 0 + self.lastPos = pos + + # ttl sweep + if rnd % 16 == 0: + self.ttl = {k: v for k, v in self.ttl.items() if v > rnd} + + # death memory (Ijti fix): ban tiles where our turrets died + if mi.turret_losses: + for (lx, ly) in mi.turret_losses: + self.ttl[('plant', lx, ly)] = rnd + 80 + mi.turret_losses.clear() + self._plantCache = None + + # claims context: my bit vs other friendly bots + my_bit = mi.bit(pos.x, pos.y) + others = mi.own_bots & ~my_bit + + states = ( + (9.0, self._score_attack, self._run_attack), + (8.0, self._score_heal, self._run_heal), + (7.75, self._score_route, self._run_route), + (7.5, self._score_secure, self._run_secure), + (4.0, self._score_harvest, self._run_harvest), + (2.0, self._score_disrupt, self._run_disrupt), + (1.0, self._score_explore, self._run_explore), + ) + best, bestScore, bestData = None, 0.0, None + for ceil, scorefn, runfn in states: + if bestScore >= ceil: + break + if ct.get_cpu_time_elapsed() > CPU_SOFT_STOP and best is not None: + break + s, data = scorefn(ct, mi, my_bit, others) + if s > bestScore: + best, bestScore, bestData = runfn, s, data + if best is not None: + best(ct, mi, bestData) + + # passive extras with leftover action/cpu + self._rebuildBroken(ct, mi) + self._tryChokeBlock(ct, mi) + self._healAdjacent(ct) + mi.share(ct, self.myNum, self.teamCore) + self._advanceAnalysis(ct) + + # ---- setup helpers ---- + def _setup(self, ct): + if self.mi is None: + w, h = ct.get_map_width(), ct.get_map_height() + self.mi = mapinfo_mod.MapInfo(w, h) + mapanalysis.configure_map(w, h) + self.mi.on_block_change = ( + lambda x, y, blocking: mapanalysis.note_structural_tile( + x, y, blocking, w, h)) + if self.myNum == -1: + self.myNum = ct.read_store(0) + 1 + return self.mi + + def _updateCores(self, ct): + if self.teamCore is None: + for bid in ct.get_nearby_buildings(): + if (ct.get_entity_type(bid) == EntityType.CORE + and ct.get_team(bid) == ct.get_team()): + self.teamCore = ct.get_position(bid) if self.enemyCore is None: - self.enemyCore = findEnemyCore(ct) - - team = ct.get_team() - - priority = { - EntityType.CORE: 5, - EntityType.SENTINEL: 3, - EntityType.GUNNER: 3, - EntityType.BARRIER: 2, - EntityType.LAUNCHER: 2, - EntityType.CONVEYOR: 1, - } - - # HARVESTER remains intentionally excluded. - targets = [ - EntityType.CORE, - EntityType.GUNNER, - EntityType.SENTINEL, - EntityType.BARRIER, - EntityType.LAUNCHER, - EntityType.CONVEYOR, - ] - - bestTile = None - bestPriority = 0 - - for tile in ct.get_attackable_tiles(): - buildingID = ct.get_tile_building_id(tile) - builderID = ct.get_tile_builder_bot_id(tile) - - # Do not attack a friendly builder. - if builderID is not None and ct.get_team(builderID) == team: - continue - - # Do not fire into a friendly building, even when an enemy - # builder is standing on it. - if buildingID is not None and ct.get_team(buildingID) == team: - continue + for bid in ct.get_nearby_buildings(): + if (ct.get_entity_type(bid) == EntityType.CORE + and ct.get_team(bid) != ct.get_team()): + self.enemyCore = ct.get_position(bid) - tilePriority = 0 - - if buildingID is not None: - buildingType = ct.get_entity_type(buildingID) + def _updateSymmetry(self, ct): + mi = self.mi + if self.teamCore is None: + return + if self.symTracker is None: + self.symTracker = symmetry.SymmetryTracker(mi.w, mi.h) + if self.fullmapVer != mi.struct_version or self.fullmap is None: + fm = [[-1] * mi.h for _ in range(mi.w)] + for x in range(mi.w): + col = fm[x] + for y in range(mi.h): + b = 1 << (x + y * mi.w) + if not (mi.seen & b): + continue + if mi.walls & b: + col[y] = 2 + elif mi.blocked & b: + col[y] = 3 + elif mi.ore & b: + col[y] = 1 + else: + col[y] = 0 + self.fullmap = fm + self.fullmapVer = mi.struct_version + self.symTracker.update(self.fullmap, mi.struct_version, self.teamCore) - # Preserves the intentional harvester exclusion. - if buildingType not in targets: + def _enemyCoreGuess(self): + if self.enemyCore is not None: + return self.enemyCore + if self.symTracker is not None and self.teamCore is not None: + cx, cy = self.symTracker.enemy_core(self.teamCore) + if 0 <= cx < self.mi.w - 1 and 0 <= cy < self.mi.h - 1: + return Position(cx, cy) + return None + + def _homeMask(self): + """Tiles within squared distance 100 of the core (Pantheon: the very + first builder restricts all its targets to this disc = home guard).""" + mi = self.mi + if self._homeDisc is not None or self.teamCore is None: + return self._homeDisc or 0 + cx, cy = self.teamCore.x, self.teamCore.y + m = 0 + for x in range(max(0, cx - 10), min(mi.w, cx + 11)): + for y in range(max(0, cy - 10), min(mi.h, cy + 11)): + if (x - cx) ** 2 + (y - cy) ** 2 <= 100: + m |= mi.bit(x, y) + self._homeDisc = m + return m + + def _coreMask(self): + mi = self.mi + if self.teamCore is None: + return 0 + m = 0 + for dx in (0, 1): + for dy in (0, 1): + x, y = self.teamCore.x + dx, self.teamCore.y + dy + if 0 <= x < mi.w and 0 <= y < mi.h: + m |= mi.bit(x, y) + return m + + # ---- movement ---- + def _moveToward(self, ct, targets_mask, avoid=0): + if ct.get_move_cooldown() != 0: + return + mi = self.mi + pos = ct.get_position() + step = path.next_step(mi, (pos.x, pos.y), targets_mask, avoid) + if step is not None: + d = DELTA_TO_DIR[step] + dest = pos.add(d) + # break-walk own barriers (Pantheon): destroy is free, remember + # the tile so we rebuild it once we've stepped off. + if (self._inBounds(dest) + and mi.own_barriers & mi.bit(dest.x, dest.y) + and not ct.can_move(d) + and ct.is_in_vision(dest) and ct.can_destroy(dest)): + ct.destroy(dest) + mi.own_barriers &= ~mi.bit(dest.x, dest.y) + mi.note_tile(dest.x, dest.y, mapinfo_mod.T_EMPTY) + if len(self.brokenBarriers) < 4: + self.brokenBarriers.append(dest) + if ct.can_move(d): + ct.move(d) + self.stuck = 0 + return + self.stuck += 1 + if self.stuck > 2 + (ct.get_id() % 8): + dirs = [d for d in CARDINALS if ct.can_move(d)] + if dirs: + ct.move(random.choice(dirs)) + + # ------------------------------------------------------------------ + # STATE: attack (9) — plant turrets on enemy buildings / push their core + # ------------------------------------------------------------------ + def _score_attack(self, ct, mi, my_bit, others): + rnd = ct.get_current_round() + ti = ct.get_global_resources() + if ti < ct.get_gunner_cost() + 40: + return 0.0, None + # DEFENSIVE counter-battery: enemy building near OUR core — every + # builder may respond (no role gate), claims still deconflict. + if self.teamCore is not None: + home = 0 + for (x, y) in mi.enemy_buildings: + if self.ttl.get(('atk', x, y), 0) > rnd: continue - - tilePriority = priority.get(buildingType, 1) - - # Preserve the existing bonus for an enemy builder - # standing on a targetable enemy building. - if builderID is not None: - tilePriority += 2 - - elif builderID is not None: - # the part that got cooked by increasing prio to 2 - continue - - if tilePriority > bestPriority: - bestPriority = tilePriority - bestTile = tile - - # Fire only after examining every attackable tile. - if bestTile is not None and ct.can_fire(bestTile): - ct.fire(bestTile) - - def launcher(self, ct: Controller): - myLoc = ct.get_position() - - for d in Directions: - adjacent = myLoc.add(d) - if adjacent.x < 0 or adjacent.y < 0 or adjacent.x >= ct.get_map_width() or adjacent.y >= ct.get_map_height(): + if abs(x - self.teamCore.x) + abs(y - self.teamCore.y) <= 10: + home |= mi.bit(x, y) + if home: + mine = path.claims(mi, my_bit, others, home) + if mine: + return 9.0, ('bld', mine) + if rnd < 30 or self.myNum % 3 != 0 or self.myNum == 1: + return 0.0, None # builder #1 is the home guard, never sieges + guess = self._enemyCoreGuess() + # bit-sliced plane placement (Pantheon): score EVERY legal gunner + # placement board-wide, take the best above threshold. + plant = self._bestPlant(ct, mi) + if plant is not None: + score, px, py, fk = plant + # geographic money gate on the PLACEMENT tile + if self.teamCore is not None and guess is not None: + dHome = abs(px - self.teamCore.x) + abs(py - self.teamCore.y) + dEnemy = abs(px - guess.x) + abs(py - guess.y) + deep = dHome > dEnemy + else: + deep = False + if not deep or ti >= 150: + mine = path.claims(mi, my_bit, others, mi.bit(px, py)) + if mine: + return 9.0, ('plant', (px, py, fk)) + # fallback: walk at remembered enemy buildings (old greedy pathway) + cands = 0 + for (x, y) in mi.enemy_buildings: + key = ('atk', x, y) + if self.ttl.get(key, 0) > rnd: continue - bot_id = ct.get_tile_builder_bot_id(adjacent) - if bot_id is not None and ct.get_team(bot_id) != ct.get_team(): + if self.teamCore is not None and guess is not None: + dHome = abs(x - self.teamCore.x) + abs(y - self.teamCore.y) + dEnemy = abs(x - guess.x) + abs(y - guess.y) + onOurHalf = dHome <= dEnemy + else: + onOurHalf = True + if onOurHalf or ti >= 150: + cands |= mi.bit(x, y) + if cands: + mine = path.claims(mi, my_bit, others, cands) + if mine: + return 8.9, ('bld', mine) + return 0.0, None + if guess is not None: + return 6.0, ('core', guess) + return 0.0, None + + PLANT_VALUES = None # built lazily: EntityType -> placement value + + def _bestPlant(self, ct, mi): + """Best gunner placement from the score planes, cached per + (struct, enemy-set). Threshold 20 = better than a lone barrier.""" + if Player.PLANT_VALUES is None: + Player.PLANT_VALUES = { + EntityType.CORE: 60, EntityType.GUNNER: 40, + EntityType.SENTINEL: 40, EntityType.LAUNCHER: 20, + EntityType.CONVEYOR: 15, EntityType.SPLITTER: 18, + EntityType.HARVESTER: 8, EntityType.BARRIER: 5, + } + key = (mi.struct_version, + tuple(sorted((k, str(v)) + for k, v in mi.enemy_buildings.items()))) + if self._plantCache is not None and self._plantCache[0] == key: + return self._plantCache[1] + by_value = {} + for (x, y), etype in mi.enemy_buildings.items(): + v = Player.PLANT_VALUES.get(etype, 5) + by_value.setdefault(v, [0])[0] = by_value.get(v, [0])[0] \ + | mi.bit(x, y) + class_masks = [(v, m[0]) for v, m in by_value.items()] + valid = (mi.seen & ~mi.walls & ~mi.blocked & ~mi.ore + & ~mi.own_conveyors & ~mi.own_barriers + & ~mi.expand(mi.expand(mi.enemy_bots))) + # death memory (Ijti fix): freshly-fatal tiles are banned + rnd = ct.get_current_round() + for (kind, kx, ky), exp in self.ttl.items(): + if kind == 'plant' and exp > rnd: + valid &= ~mi.bit(kx, ky) + best = None + if class_masks and valid: + # threat penalty (Ijti fix): unthreatened placements first; + # threatened only at 2x value (trade, not feeding tube) + threat = mi.threat() + best = planes.best_placement(mi, class_masks, valid & ~threat) + if best is None or best[0] < 20: + hot = planes.best_placement(mi, class_masks, valid & threat) + if hot is not None and hot[0] >= 40 \ + and (best is None or hot[0] > best[0]): + best = hot + if best is not None and best[0] < 20: best = None - bestDist = 0 - for tile in ct.get_nearby_tiles(26): - if ct.can_launch(adjacent, tile): - dist = tile.distance_squared(myLoc) - if dist > bestDist: - bestDist = dist - best = tile - if best is not None: - ct.launch(adjacent, best) + self._plantCache = (key, best) + return best + + def _enemyFieldCached(self, mi, guess): + key = (mi.struct_version, guess.x, guess.y) + if self.enemyField is not None and self.enemyFieldVer == key: + return self.enemyField + m = 0 + for dx in (0, 1): + for dy in (0, 1): + x, y = guess.x + dx, guess.y + dy + if 0 <= x < mi.w and 0 <= y < mi.h: + m |= mi.bit(x, y) + self.enemyField = path.dist_field(mi, mi.expand(m)) + self.enemyFieldVer = key + return self.enemyField + + def _run_attack(self, ct, mi, data): + kind, payload = data + pos = ct.get_position() + rnd = ct.get_current_round() + if kind == 'core': + # ATTACK ROUTE (Pantheon override): while marching on the enemy + # core, lay a conveyor trail once we're geographically ahead + # (dist-to-them <= 1.5x dist-home). Walkable for us, area denial + # for them; planes convert the head into a gunner on contact. + if (ct.get_action_cooldown() == 0 + and ct.get_global_resources() + >= ct.get_conveyor_cost() + 120): + ef = self._enemyFieldCached(mi, payload) + hf = self._routeFieldCached(ct) + idx = pos.x + pos.y * mi.w + if (ef is not None and hf is not None + and ef[idx] < 4096 and hf[idx] < 4096 + and ef[idx] <= 1.5 * hf[idx]): + best = None + for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)): + nx, ny = pos.x + dx, pos.y + dy + if not (0 <= nx < mi.w and 0 <= ny < mi.h): + continue + b = mi.bit(nx, ny) + if (mi.walls | mi.blocked | mi.own_conveyors + | mi.own_barriers | mi.ore) & b: + continue + d = ef[nx + ny * mi.w] + if best is None or d < best[0]: + best = (d, nx, ny) + if best is not None and best[0] < ef[idx]: + _, lx, ly = best + link = Position(lx, ly) + # face further downhill toward the enemy core + fb = None + for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)): + fx, fy = lx + dx, ly + dy + if not (0 <= fx < mi.w and 0 <= fy < mi.h): + continue + fd = ef[fx + fy * mi.w] + if fb is None or fd < fb[0]: + fb = (fd, dx, dy) + if fb is not None: + facing = DELTA_TO_DIR[(fb[1], fb[2])] + if (ct.is_in_vision(link) + and ct.can_build_conveyor(link, facing)): + ct.build_conveyor(link, facing) + mi.own_conveyors |= mi.bit(lx, ly) + mi.own_conv_facing[(lx, ly)] = (fb[1], fb[2]) + m = mi.bit(payload.x, payload.y) + self._moveToward(ct, mi.expand(mi.expand(m))) + return + if kind == 'plant': + px, py, fk = payload + tile = Position(px, py) + facing = {'E': Direction.EAST, 'W': Direction.WEST, + 'N': Direction.NORTH, 'S': Direction.SOUTH}[fk] + tb = mi.bit(px, py) + if pos == tile: + # standing on the placement: step aside + self._moveToward(ct, mi.expand(tb) & mi.passable() & ~tb) + return + if pos.distance_squared(tile) <= 2 \ + and ct.get_action_cooldown() == 0: + if ct.is_in_vision(tile) and ct.can_build_gunner(tile, facing): + ct.build_gunner(tile, facing) + self._plantCache = None return - - def builderBot(self, ct: Controller) -> None: - myLoc = ct.get_position() - if ct.get_hp() < ct.get_max_hp(): - if ct.can_heal(myLoc): - ct.heal(myLoc) - currentRound = ct.get_current_round() - self.setUp(ct, myLoc) - if ct.get_id() < 5 and ((self.currentHarvester is None and (currentRound > 20 or self.nearCore2(myLoc) < 4))): - self.alwaysDefense = True - self.defenseMode = True - if currentRound % 120 == 1 or self.justSpawned: - self.assignRole(ct, currentRound) - self.justSpawned = False - self.defender.defendInfrastructure(ct, self.teamCore, self.pf) - - if self.attackMode == True: - self.attack(ct) + self.ttl[('atk', px, py)] = rnd + 20 + self._plantCache = None + return + self._moveToward(ct, mi.expand(tb) & mi.passable()) return - if self.defenseMode: - self.defender.defendInfrastructure(ct, self.teamCore, self.pf) - healPos = self.defender.picHealingLoc(ct, self.teamCore, self.alwaysDefense) - if self.nearCore2(myLoc) < 4 and ct.get_global_resources() > 40: - if ct.can_build_conveyor(myLoc, myLoc.direction_to(self.teamCore)): - ct.build_conveyor(myLoc, myLoc.direction_to(self.teamCore)) - if healPos is not None: - self.pf.moveTo(ct, healPos) + targets = payload + tb = targets & -targets + tx, ty = mi.xy(tb) + tpos = Position(tx, ty) + dist = manhattan(pos, tpos) + if dist <= 3 and ct.get_action_cooldown() == 0: + # place a turret on an adjacent tile with line to the target + facing = self._cardinalTo(pos, tpos) + placed = False + for d in CARDINALS: + p = pos.add(d) + if not self._inBounds(p) or not ct.is_in_vision(p): + continue + f = self._cardinalTo(p, tpos) + if f == Direction.CENTRE: + continue + if ct.can_build_gunner(p, f): + ct.build_gunner(p, f) + placed = True + break + if (ct.get_global_resources() >= ct.get_sentinel_cost() + 40 + and ct.can_build_sentinel(p, f)): + ct.build_sentinel(p, f) + placed = True + break + if placed: + self.ttl[('atk', tx, ty)] = rnd + 40 + return + if facing == Direction.CENTRE: + self.ttl[('atk', tx, ty)] = rnd + TTL_ROUNDS + return + if dist <= 1: + # cannot place and already adjacent: back off this target a while + self.ttl[('atk', tx, ty)] = rnd + 30 + self._moveToward(ct, mi.expand(targets) & mi.passable()) + + # ------------------------------------------------------------------ + # STATE: heal (8) — repair damaged own buildings, deter intruders + # ------------------------------------------------------------------ + def _score_heal(self, ct, mi, my_bit, others): + damaged = 0 + # core distress broadcast: converge on the core even when it's out of + # our own vision — 2 medics out-heal a sentinel's grind (8 vs 6/round). + distress = ct.read_store(7) + if (distress > 0 and ct.get_current_round() - distress < 12 + and self.teamCore is not None): + damaged |= self._coreMask() + for bid in ct.get_nearby_buildings(): + if ct.get_team(bid) != ct.get_team(): + continue + if ct.get_hp(bid) + 2 < ct.get_max_hp(bid): + p = ct.get_position(bid) + damaged |= mi.bit(p.x, p.y) + if damaged: + mine = path.claims(mi, my_bit, others, damaged) + if mine: + return 8.0, ('heal', mine) + # Pantheon: own sentinel trading with an enemy sentinel counts as very + # damaged — medics pre-park. Scores 7, BELOW route's 7.75 (the paper's + # exact number; 8.0 parked the workforce and starved the economy). + trading = 0 + for (x, y), tk in mi.own_turrets.items(): + if tk != 'S': + continue + for (ex, ey), ekind in mi.enemy_turrets.items(): + if ekind == 'S' and (x - ex) ** 2 + (y - ey) ** 2 <= 32: + trading |= mi.bit(x, y) + break + if trading: + mine = path.claims(mi, my_bit, others, trading) + if mine: + return 7.0, ('heal', mine) + # Pantheon chase zone: 8 pathing steps around our conveyor network + # (was 2 tiles) — intruders get hunted before they reach the chain. + zone = mi.own_conveyors + passable = mi.passable() + for _ in range(8): + zone = mi.expand(zone) & (passable | zone) + intruders = mi.enemy_bots & zone + if intruders: + mine = path.claims(mi, my_bit, others, intruders) + if mine: + return 7.9, ('chase', mine) + return 0.0, None + + def _run_heal(self, ct, mi, data): + kind, mask = data + pos = ct.get_position() + tb = mask & -mask + tx, ty = mi.xy(tb) + tpos = Position(tx, ty) + if kind == 'heal': + if pos.distance_squared(tpos) <= 2: + if ct.can_heal(tpos): + ct.heal(tpos) + return + self._moveToward(ct, mi.expand(tb) & mi.passable()) return - else: - if self.currentHarvester is not None: - if self.surroundHarvester: - self.buildConveyor(ct) - else: - self.protectHarvester(ct) - else: - self.getHarvester(ct, myLoc) - if self.foundHarvester is None and self.currentHarvester is None: - if self.explorePos is None or self.exploreTime > 35 or self.explorePos.distance_squared(myLoc) <= 4: - self.pickExplore(ct, currentRound) - self.exploreTime = 0 - self.exploreTime += 1 - ct.draw_indicator_line(myLoc, self.explorePos, 255, 255, 255) - self.pf.moveTo(ct, self.explorePos) - - def setUp(self, ct: Controller, myLoc: Position): + # chase: stand next to the intruder; the launcher/turret net does the + # rest, and our body denies conveyor-chip tiles + if pos.distance_squared(tpos) <= 2: + if ct.get_action_cooldown() == 0: + for d in CARDINALS: + p = pos.add(d) + if (self._inBounds(p) and ct.is_in_vision(p) + and manhattan(p, tpos) <= 1 + and ct.can_build_launcher(p) + and ct.get_global_resources() + >= ct.get_launcher_cost() + 60): + ct.build_launcher(p) + return + return + # Pantheon counter-launcher fallback: chase can't PATH to the + # intruder (their launcher zone / barriers) -> zone them back + # instead of giving up. + step = path.next_step(mi, (pos.x, pos.y), + mi.expand(tb) & mi.passable()) + if step is None: + if (manhattan(pos, tpos) <= 3 and ct.get_action_cooldown() == 0 + and ct.get_global_resources() + >= ct.get_launcher_cost() + 60): + for d in CARDINALS: + p = pos.add(d) + if (self._inBounds(p) and ct.is_in_vision(p) + and manhattan(p, tpos) <= 2 + and ct.can_build_launcher(p)): + ct.build_launcher(p) + return + self._moveToward(ct, mi.expand(mi.expand(tb)) & mi.passable()) + return + self._moveToward(ct, mi.expand(tb) & mi.passable()) + + # ------------------------------------------------------------------ + # STATE: route (7.75) — stateless chain building down the core gradient + # ------------------------------------------------------------------ + def _connectedConveyors(self, mi): + """Own conveyors whose chain (by facing) reaches the core — legal + merge targets for new routes (Pantheon: routes may end on existing + conveyors). Fixpoint over remembered facings.""" + coreMask = self._coreMask() + core_xy = set() + m = coreMask + while m: + lsb = m & -m + core_xy.add(mi.xy(lsb)) + m ^= lsb + connected = set() + changed = True + while changed: + changed = False + for (x, y), (dx, dy) in mi.own_conv_facing.items(): + if (x, y) in connected: + continue + out = (x + dx, y + dy) + if out in core_xy or out in connected: + connected.add((x, y)) + changed = True + return connected + + def _routeFieldCached(self, ct): + mi = self.mi + key = (mi.struct_version, len(mi.own_conv_facing), + tuple(sorted(mi.enemy_turrets))) + if self.routeField is not None and self.routeVer == key: + return self.routeField + coreMask = self._coreMask() + if not coreMask: + return None + # Pantheon merge targets: the core PLUS every conveyor already + # connected to it — new chains join existing trunks instead of each + # harvester paying for a private line the whole way home. LOADED + # trunk tiles are excluded (P6): a full conveyor can't accept a new + # stack, so merging there jams the junction (their +4 penalty, made + # binary by dev23's hold-1-stack rule). + self._connected = self._connectedConveyors(mi) + targets = coreMask + for (x, y) in self._connected: + targets |= mi.bit(x, y) + targets &= ~mi.own_loaded | coreMask + # Pantheon universal AVOID mask (p7): tiles covered by enemy turrets + # are HARD-excluded from conveyor route planning, not soft-costed — + # a link placed under a known gun is titanium fed to the enemy + # (Landers vault r139: we rebuilt one threatened link 7 times). + # Existing own conveyors stay routable so trunks aren't orphaned. + avoid = mi.threat() & ~mi.own_conveyors & ~coreMask + self.routeField = path.dist_field(mi, targets, avoid_mask=avoid) + self.routeVer = key + return self.routeField + + @staticmethod + def _chainRoot(mi, hx, hy): + """Adjacent own conveyor whose output is NOT the harvester — a real + chain root. Inward-facing guards don't count (Khaos guard mask).""" + for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)): + cx, cy = hx + dx, hy + dy + f = mi.own_conv_facing.get((cx, cy)) + if f is None: + continue + if (cx + f[0], cy + f[1]) != (hx, hy): + return (cx, cy) + return None + + def _downhillLink(self, mi, field, x, y): + """Best empty neighbour to lay the next conveyor on, plus the facing + that conveyor should have (toward ITS best downhill neighbour).""" + best = None + for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)): + nx, ny = x + dx, y + dy + if not (0 <= nx < mi.w and 0 <= ny < mi.h): + continue + b = mi.bit(nx, ny) + if (mi.walls | mi.blocked | mi.own_conveyors) & b: + continue + d = field[nx + ny * mi.w] + if best is None or d < best[0]: + best = (d, nx, ny) + if best is None or best[0] >= 4096: + return None, None + _, lx, ly = best + fbest = None + for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)): + fx, fy = lx + dx, ly + dy + if not (0 <= fx < mi.w and 0 <= fy < mi.h): + continue + fd = field[fx + fy * mi.w] + if fbest is None or fd < fbest[0]: + fbest = (fd, dx, dy) + if fbest is None: + return None, None + return Position(lx, ly), DELTA_TO_DIR[(fbest[1], fbest[2])] + + def _orphanHarvesters(self, ct, mi): + rnd = ct.get_current_round() + orphans = 0 + for (x, y) in mi.own_harvesters: + key = ('route', x, y) + if self.ttl.get(key, 0) > rnd: + continue + if self._chainRoot(mi, x, y) is None: + orphans |= mi.bit(x, y) + return orphans + + def _deadEnds(self, ct, mi): + """Own conveyors whose output tile has no continuation (Khaos + dead-end route targets). Guards (facing into a harvester) excluded.""" + rnd = ct.get_current_round() + coreMask = self._coreMask() + ends = 0 + for (x, y), (dx, dy) in mi.own_conv_facing.items(): + if self.ttl.get(('route', x, y), 0) > rnd: + continue + ox, oy = x + dx, y + dy + if (ox, oy) in mi.own_harvesters: + continue # inward guard, carries nothing + if not (0 <= ox < mi.w and 0 <= oy < mi.h): + ends |= mi.bit(x, y) + continue + ob = mi.bit(ox, oy) + if ob & coreMask: + continue # delivers into the core: chain complete + if mi.own_conveyors & ob: + continue # continues into another conveyor/splitter + ends |= mi.bit(x, y) + return ends + + def _score_route(self, ct, mi, my_bit, others): if self.teamCore is None: - self.teamCore = findTeamCore(ct) - if self.teamCore is not None: - self.teamCoreTiles = self._computeCoreTiles(ct, self.teamCore) - if self.enemyCore is None: - self.enemyCore = findEnemyCore(ct) - if self.oldPos is not None and myLoc.distance_squared(self.oldPos) > 2: - self.pf.reset() - self.oldPos = myLoc - - def _computeCoreTiles(self, ct: Controller, corner: Position) -> list: - cid = ct.get_tile_building_id(corner) - tiles = [] - if cid is not None: - for dx in (-1, 0, 1): - for dy in (-1, 0, 1): - q = Position(corner.x + dx, corner.y + dy) - if ct.is_in_vision(q) and self.isInBounds(ct, q) and ct.get_tile_building_id(q) == cid: - tiles.append(q) - return tiles if tiles else [corner] - - def nearCore2(self, pos: Position) -> int: - tiles = self.teamCoreTiles if self.teamCoreTiles else [self.teamCore] - return min(pos.distance_squared(t) for t in tiles) - - def assignRole(self, ct: Controller, currentRound): - - if currentRound <= 120: - # 36% Offense, 64% Eco + 1 Defense Bot - if currentRound % 11 in [3, 5, 7, 10] and self.currentHarvester is None: - self.attackMode = True - else: - self.defenseMode = False - self.attackMode = False - if currentRound % 120 == 1 and currentRound > 120: - # 25% Defense, 35% Offense, 40% Eco - self.defenseMode = False - self.attackMode = False - if self.currentHarvester is None: - role = random.random() - if role > 0.75 or self.alwaysDefense == True: - self.defenseMode = True - elif role > 0.4 and currentRound > 120: - self.attackMode = True - - def attack(self, ct: Controller) -> None: - myLoc = ct.get_position() - - coreHypothesis = [ - #rotational sym - Position( - ct.get_map_width() - 2 - self.teamCore.x, - ct.get_map_height() - 2 - self.teamCore.y - ), - - # vertical axis reflc - Position( - ct.get_map_width() - 2 - self.teamCore.x, - self.teamCore.y - ), - - # horizontal axis reflc - Position( - self.teamCore.x, - ct.get_map_height() - 2 - self.teamCore.y - ), - ] - # an observed enemy core overrides the new hypotheses - if self.enemyCore is not None: - self.attackPos = self.enemyCore - # Determine attack destination - elif self.attackPos is None: - self.attackPos = coreHypothesis[self.attackHypothesisIndex] - self.attackMoveTimeout = 0 - - def nextAttackHypothesis(): - self.attackHypothesisIndex = ( - self.attackHypothesisIndex + 1 - ) % len(coreHypothesis) - self.attackPos = coreHypothesis[self.attackHypothesisIndex] - self.attackMoveTimeout = 0 - - def changeAttackPos (ct: Controller): - dx = 5 * random.randint(-1, 1) # more consistently farther away - dy = 5 * random.randint(-1, 1) - new_x = max(0, min(self.attackPos.x + dx, ct.get_map_width() - 1)) - new_y = max(0, min(self.attackPos.y + dy, ct.get_map_height() - 1)) - - self.attackPos = Position(new_x, new_y) - self.attackMoveTimeout = 0 - - # Jitter so we don't stand still on the target - if myLoc.distance_squared(self.attackPos) < 8: - changeAttackPos(ct) - - - if self.spawnSentinelHere is None: # Pick a sentinel placement spot near the attack target - nearbyEnemyStuff = [] - isStuff = False - - for i in ct.get_nearby_buildings(): - if ct.get_entity_type(i) == EntityType.CONVEYOR: - if ct.get_team(i) != ct.get_team() and ct.get_stored_resource(i) is not None: - nearbyEnemyStuff.append(ct.get_position(i)) - isStuff = True - nearbyEnemyStuff.sort(key=lambda p: p.distance_squared(self.attackPos)) - - if not isStuff: # no enemy infastructure was found - self.attackMoveTimeout += 1 - self.pf.moveTo(ct, self.attackPos) - if self.attackMoveTimeout > 20: # While moving to attack pos, no enemy infastructure was found for 20 turns, change it - if self.enemyCore is None: - nextAttackHypothesis() - else: - changeAttackPos(ct) - return - else: # stuff was found, now reuse attackMoveTimeout for sentinelSpot - self.spawnSentinelHere = nearbyEnemyStuff[0] - self.attackMoveTimeout = 0 - - # From now on there must be a sentinel Spot - - # first check if there is already another sentinel close by - for b_id in ct.get_nearby_buildings(): - if ct.get_entity_type(b_id) == EntityType.SENTINEL and ct.get_team(b_id) == ct.get_team(): - if ct.get_position(b_id).distance_squared(self.spawnSentinelHere) <= 8: - self.spawnSentinelHere = None - self.attackMoveTimeout = 0 - self.pf.moveTo(ct, self.attackPos) - return - - if self.attackMoveTimeout > 25: # This spot is hard to get, switch to another one - self.spawnSentinelHere = None - self.attackMoveTimeout = 0 + return 0.0, None + cands = self._orphanHarvesters(ct, mi) | self._deadEnds(ct, mi) + if not cands: + return 0.0, None + mine = path.claims(mi, my_bit, others, cands) + if not mine: + return 0.0, None + near_enemy = mi.expand(mi.expand(mi.enemy_bots)) + important = mine & near_enemy + return (7.75 if important else 5.0), mine + + def _run_route(self, ct, mi, mask): + field = self._routeFieldCached(ct) + rnd = ct.get_current_round() + hb = mask & -mask + hx, hy = mi.xy(hb) + if field is None: return - - if myLoc.distance_squared(self.spawnSentinelHere) < 1: # this means you are close enough to the sentinelSpot - self.attackMoveTimeout = 0 + coreMask = self._coreMask() + link = facing = None + if (hx, hy) in mi.own_conv_facing: + root = (hx, hy) # dead-end conveyor target: walk from it else: - self.pf.moveTo(ct, self.spawnSentinelHere) - self.attackMoveTimeout += 1 - return - - - if ct.get_tile_building_id(self.spawnSentinelHere) is not None: # if there is something else there clear it - thingTeam = ct.get_team(ct.get_tile_building_id(self.spawnSentinelHere)) - if thingTeam == ct.get_team(): - if ct.can_destroy(self.spawnSentinelHere): - ct.destroy(self.spawnSentinelHere) - else: - self.defender.placeLauncher(ct, self.spawnSentinelHere) - self.pf.moveTo(ct, self.spawnSentinelHere) - if ct.can_fire(self.spawnSentinelHere) and ct.get_global_resources() > 25: - ct.fire(self.spawnSentinelHere) - - - if ct.get_tile_building_id(self.spawnSentinelHere) is None: # Tile is clear — build sentinel facing the most valuable targets - if myLoc == self.spawnSentinelHere: - self.pf.moveTo(ct, self.attackPos) - directionChoices = {d: 0 for d in DIRECTIONS} - for i in ct.get_nearby_buildings(): - if ct.get_team(i) != ct.get_team(): - d = self.spawnSentinelHere.direction_to(ct.get_position(i)) - if ct.get_entity_type(i) == EntityType.CORE: - directionChoices[d] += 25 - else: - directionChoices[d] += 1 - - directionChoice = max(directionChoices, key=lambda d: directionChoices[d]) - - if ct.can_build_sentinel(self.spawnSentinelHere, directionChoice): - ct.build_sentinel(self.spawnSentinelHere, directionChoice) - self.spawnSentinelHere = None - self.attackPos = None - - - - def pickExplore(self, ct: Controller, currentRound): - w = ct.get_map_width() - h = ct.get_map_height() - myLoc = ct.get_position() - round = ct.get_current_round() - radius = (w + h) // 5 - if round > 5: - theta = random.random() * 2 * math.pi + root = self._chainRoot(mi, hx, hy) + # P6 cost gate (Pantheon cost(d)): before STARTING a fresh chain, + # check we can afford ~the whole line. 0.65 discount — income + # keeps arriving while we build. + d = field[hx + hy * mi.w] + if d < 4096: + est = 3 * d * ct.get_scale_percent() / 100 * 0.65 + if ct.get_global_resources() < est: + self.ttl[('route', hx, hy)] = rnd + 30 + return + if root is None: + link, facing = self._downhillLink(mi, field, hx, hy) else: - theta = (((3 * round) % 5) / 5) * 2 * math.pi - tx = int(max(0, min(w - 1, myLoc.x + math.cos(theta) * radius))) - ty = int(max(0, min(h - 1, myLoc.y + math.sin(theta) * radius))) - self.explorePos = Position(tx, ty) - - def getHarvester(self, ct: Controller, myLoc: Position): - if self.foundHarvester is None: - nearbyOres = [] - for i in ct.get_nearby_tiles(): - if ct.is_in_vision(i): - if ct.get_tile_env(i) == Environment.ORE_TITANIUM: - nearbyOres.append(i) - nearbyOres.sort(key=lambda p: p.distance_squared(myLoc) + self.nearCore2(p)) - found = False - for t in nearbyOres: - tID = ct.get_tile_building_id(t) - willWork = False - if t not in self.usedHarvesters: - if tID is None: - willWork = True - elif ct.get_team(tID) == ct.get_team(): - willWork = True - elif ct.get_entity_type(tID) == EntityType.HARVESTER: - willWork = True - if willWork and self.foundHarvester is None: - self.foundHarvester = t - self.reachHarvesterTime = 0 - found = True - if found == False: + # follow the actual chain by facings until it breaks / reaches core + cur = root + visited = set() + for _ in range(mi.w + mi.h): + if cur in visited: + self.ttl[('route', hx, hy)] = rnd + 30 + return + visited.add(cur) + f = mi.own_conv_facing.get(cur) + if f is None: + self.ttl[('route', hx, hy)] = rnd + 15 + return + ox, oy = cur[0] + f[0], cur[1] + f[1] + if not (0 <= ox < mi.w and 0 <= oy < mi.h): + link, facing = self._downhillLink(mi, field, + cur[0], cur[1]) + break + ob = mi.bit(ox, oy) + if ob & coreMask or (ox, oy) in self._connected: + self.ttl[('route', hx, hy)] = rnd + 30 # chain complete + return + if (ox, oy) in mi.own_conv_facing: + cur = (ox, oy) + continue + if mi.own_conveyors & ob: + # splitter / facing not yet observed: assume connected + self.ttl[('route', hx, hy)] = rnd + 15 + return + if (mi.walls | mi.blocked) & ob: + link, facing = self._downhillLink(mi, field, + cur[0], cur[1]) + break + # empty output tile: that's the missing link + link = Position(ox, oy) + fbest = None + for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)): + fx, fy = ox + dx, oy + dy + if not (0 <= fx < mi.w and 0 <= fy < mi.h): + continue + fd = field[fx + fy * mi.w] + if fbest is None or fd < fbest[0]: + fbest = (fd, dx, dy) + if fbest is None: + return + facing = DELTA_TO_DIR[(fbest[1], fbest[2])] + break + if link is None: + # Pantheon: an unroutable dead-end near enemies gets CAPPED with + # a barrier on its output so they can't exploit/extend it + f = mi.own_conv_facing.get((hx, hy)) + if f is not None and mi.enemy_bots: + near = mi.expand(mi.expand(mi.expand(mi.expand( + mi.bit(hx, hy))))) + if near & mi.enemy_bots and ct.get_action_cooldown() == 0: + ox, oy = hx + f[0], hy + f[1] + if 0 <= ox < mi.w and 0 <= oy < mi.h: + cap = Position(ox, oy) + pos0 = ct.get_position() + if (pos0.distance_squared(cap) <= 2 + and ct.is_in_vision(cap) + and ct.can_build_barrier(cap)): + ct.build_barrier(cap) + mi.own_barriers |= mi.bit(ox, oy) + mi.note_tile(ox, oy, mapinfo_mod.T_BLOCK) + self.ttl[('route', hx, hy)] = rnd + 20 + return + pos = ct.get_position() + lb = mi.bit(link.x, link.y) + if pos == link: + # standing on the link tile: step aside so we can build it + self._moveToward(ct, mi.expand(lb) & mi.passable() & ~lb) + return + if pos.distance_squared(link) <= 2 and ct.get_action_cooldown() == 0: + if ct.is_in_vision(link) and ct.can_build_conveyor(link, facing): + ct.build_conveyor(link, facing) + mi.own_conveyors |= lb + mi.own_conv_facing[(link.x, link.y)] = ( + mapinfo_mod.DIR_DELTA[facing]) return - if self.reachHarvesterTime >= 20: # Limit time spend trying to reach the harvester - self.removeHarvester(ct) + self.ttl[('route', hx, hy)] = rnd + 10 return - if ct.is_in_vision(self.foundHarvester) and ct.get_tile_building_id(self.foundHarvester) is not None: # Checks on current ore tile - oreID = ct.get_tile_building_id(self.foundHarvester) - if ct.get_team(oreID) == ct.get_team(): - if ct.get_entity_type(oreID) == EntityType.HARVESTER: # Already is a team harvester - self.removeHarvester(ct) + self._moveToward(ct, mi.expand(lb) & mi.passable()) + + # ------------------------------------------------------------------ + # STATE: secure (7.5) — guard-conveyor exposed harvesters near enemies + # ------------------------------------------------------------------ + def _score_secure(self, ct, mi, my_bit, others): + if not mi.enemy_buildings and not mi.enemy_bots: + return 0.0, None + danger = 0 + for (x, y) in mi.enemy_buildings: + danger |= mi.bit(x, y) + danger = mi.expand(mi.expand(mi.expand(danger))) | \ + mi.expand(mi.expand(mi.enemy_bots)) + exposed = 0 + for (x, y) in mi.own_harvesters: + b = mi.bit(x, y) + if not (danger & mi.expand(b)): + continue + open_sides = mi.expand(b) & ~b & mi.passable() & ~mi.own_conveyors + if open_sides: + exposed |= open_sides + # SECURE THE CORE (Pantheon secure-state on the ultimate exposed + # building): a conveyor ring denies point-blank turret plants AND + # absorbs distant gunner rays (ray stops at first targetable + # building — 20hp conveyor tanks it, not the 500hp core). Only under + # threat near home, so peaceful econ games skip the +scale tax. + # (lastpopperian strait r136: two gunners planted at our doorstep, + # 20dmg/round vs 8/round of medic healing.) + if self.teamCore is not None: + coreMask = self._coreMask() + ring = mi.expand(coreMask) & ~coreMask # cardinal shell-1 only: + # gunner rays are cardinal, and distant rays die in shell 1 + openRing = ring & mi.passable() & ~mi.own_conveyors \ + & ~mi.own_barriers & ~mi.ore + homeThreat = danger & self._homeMask() + if openRing and homeThreat: + exposed |= openRing + if not exposed: + return 0.0, None + mine = path.claims(mi, my_bit, others, exposed) + if not mine: + return 0.0, None + return 7.5, mine + + def _run_secure(self, ct, mi, mask): + pos = ct.get_position() + tb = mask & -mask + tx, ty = mi.xy(tb) + tpos = Position(tx, ty) + if pos.distance_squared(tpos) <= 2 and ct.get_action_cooldown() == 0: + if ct.is_in_vision(tpos): + # find the harvester this side belongs to + harv = None + for d in CARDINALS: + hp = tpos.add(d) + if (self._inBounds(hp) + and (hp.x, hp.y) in mi.own_harvesters): + harv = (hp, d) + break + if harv is None: + # core-ring tile: face INTO the core (denies the plant, + # absorbs rays, and delivers titanium) + coreMask = self._coreMask() + for d in CARDINALS: + cp = tpos.add(d) + if (self._inBounds(cp) + and coreMask & mi.bit(cp.x, cp.y)): + if ct.can_build_conveyor(tpos, d): + ct.build_conveyor(tpos, d) + mi.own_conveyors |= mi.bit(tpos.x, tpos.y) + mi.own_conv_facing[(tpos.x, tpos.y)] = ( + mapinfo_mod.DIR_DELTA[d]) + return return - else: - if ct.can_destroy(self.foundHarvester) and myLoc.distance_squared(self.foundHarvester) == 1: # Remove to have room for harvester - ct.destroy(self.foundHarvester) - elif ct.get_entity_type(oreID) != EntityType.HARVESTER: # Nothing you can do - self.removeHarvester(ct) + hp, inward = harv + # Khaos: if this side is the harvester's downhill side, face + # OUTWARD so the guard doubles as the chain root; otherwise + # face INTO the harvester so it never carries resources. + facing = inward + field = self._routeFieldCached(ct) + if field is not None: + dlink, dfacing = self._downhillLink(mi, field, hp.x, hp.y) + if dlink is not None and dlink == tpos: + facing = dfacing + if ct.can_build_conveyor(tpos, facing): + ct.build_conveyor(tpos, facing) + mi.own_conveyors |= mi.bit(tpos.x, tpos.y) + mi.own_conv_facing[(tpos.x, tpos.y)] = ( + mapinfo_mod.DIR_DELTA[facing]) return - if myLoc == self.foundHarvester: - possibleSpots = [] - for i in CardDirections: - pos = myLoc.add(i) - if self.isInBounds(ct, pos) and self.pf.canMove(ct, i): - possibleSpots.append((i, pos)) - possibleSpots.sort(key=lambda x: self.nearCore2(x[1])) - if len(possibleSpots) > 0: - self.pf._move(ct, possibleSpots[0][0]) - else: - self.usedHarvesters.append(self.foundHarvester) - self.foundHarvester = None return - myLoc = ct.get_position() - if myLoc.distance_squared(self.foundHarvester) > 1: - self.pf.moveTo(ct, self.foundHarvester) - self.reachHarvesterTime += 1 - myLoc = ct.get_position() - if myLoc.distance_squared(self.foundHarvester) == 1: - if ct.can_build_harvester(self.foundHarvester): # Built the harvester - ct.build_harvester(self.foundHarvester) - self.addHarvester(ct, myLoc) - return - oreID = ct.get_tile_building_id(self.foundHarvester) - if oreID is not None and ct.get_entity_type(oreID) == EntityType.HARVESTER and ct.get_team(oreID) != ct.get_team(): # Right next to an enemy harvester - self.addHarvester(ct, myLoc) + self._moveToward(ct, mi.expand(tb) & mi.passable()) + + # ------------------------------------------------------------------ + # STATE: harvest (4) + # ------------------------------------------------------------------ + def _score_harvest(self, ct, mi, my_bit, others): + rnd = ct.get_current_round() + # exclude only ore in the CLOSE disc of an enemy turret. The full + # facing-agnostic band mask vetoed most of the ore field on turret-y + # maps (quarry: 5-10k mined vs botv2's 18k) — band-edge mining is + # viable with heal upkeep (v29 lesson). + hot = 0 + for (x, y) in mi.enemy_turrets: + hot |= mi.bit(x, y) + hot = mi.expand(mi.expand(hot)) + cands = mi.ore & mi.seen & ~mi.blocked & ~mi.walls & ~hot + pruned = 0 + for lsb in mi.iter_bits(cands): + x, y = mi.xy(lsb) + if (x, y) in mi.own_harvesters: + continue + if self.ttl.get(('harv', x, y), 0) > rnd: + continue + pruned |= lsb + if self.myNum == 1: + pruned &= self._homeMask() # home guard mines home ore only + if not pruned: + return 0.0, None + mine = path.claims(mi, my_bit, others, pruned) + if not mine: + return 0.0, None + if ct.get_global_resources() < ct.get_harvester_cost(): + return 0.5, mine + return 4.0, mine + + def _run_harvest(self, ct, mi, mask): + pos = ct.get_position() + target = path.closest(mi, mi.bit(pos.x, pos.y), mask) + if not target: + target = mask & -mask + tx, ty = mi.xy(target) + tpos = Position(tx, ty) + rnd = ct.get_current_round() + if pos == tpos: + # standing on the ore: sidestep so we can build it + self._moveToward(ct, mi.expand(target) & mi.passable() & ~target) + return + if pos.distance_squared(tpos) <= 2 and ct.get_action_cooldown() == 0: + if ct.is_in_vision(tpos) and ct.can_build_harvester(tpos): + ct.build_harvester(tpos) + mi.own_harvesters[(tx, ty)] = True return + self.ttl[('harv', tx, ty)] = rnd + 25 return - - def removeHarvester(self, ct: Controller): - self.usedHarvesters.append(self.foundHarvester) - self.foundHarvester = None - self.reachHarvesterTime = 0 - - def addHarvester(self, ct: Controller, myLoc): - # Book keeping - self.currentHarvester = self.foundHarvester - self.usedHarvesters.append(self.foundHarvester) - self.foundHarvester = None - self.conveyorEnd = myLoc - # Protect Harvester - self.surroundHarvester = False - self.isEnemyInfastructure = None - self.sentinelSpot = None - # Placing Sentinel on Offense - self.turretTimeOut = 0 - self.enemyPos = None - self.enemyType = -1 - - def protectHarvester(self, ct: Controller): - if self.protectTurns > 12: - self.protectTurns = 0 - self.surroundHarvester = True - self.isEnemyInfastructure = None - self.sentinelSpot = None + self._moveToward(ct, mi.expand(target) & mi.passable()) + + # ------------------------------------------------------------------ + # STATE: disrupt (2) — barrier enemy-half ore when rich + # ------------------------------------------------------------------ + def _score_disrupt(self, ct, mi, my_bit, others): + if self.myNum == 1: + return 0.0, None # home guard stays home + if ct.get_global_resources() < 5 * ct.get_harvester_cost(): + return 0.0, None + guess = self._enemyCoreGuess() + if guess is None or self.teamCore is None: + return 0.0, None + rnd = ct.get_current_round() + cands = 0 + for lsb in mi.iter_bits(mi.ore & mi.seen & ~mi.blocked + & ~mi.threat()): + x, y = mi.xy(lsb) + p = Position(x, y) + if manhattan(p, guess) < manhattan(p, self.teamCore) - 2: + if self.ttl.get(('dis', x, y), 0) <= rnd: + cands |= lsb + if not cands: + return 0.0, None + mine = path.claims(mi, my_bit, others, cands) + if not mine: + return 0.0, None + return 2.0, mine + + def _run_disrupt(self, ct, mi, mask): + pos = ct.get_position() + tb = mask & -mask + tx, ty = mi.xy(tb) + tpos = Position(tx, ty) + rnd = ct.get_current_round() + if pos.distance_squared(tpos) <= 2 and pos != tpos \ + and ct.get_action_cooldown() == 0: + if ct.is_in_vision(tpos) and ct.can_build_barrier(tpos): + ct.build_barrier(tpos) + self.ttl[('dis', tx, ty)] = rnd + 1000 + return + self.ttl[('dis', tx, ty)] = rnd + 50 return - - self.protectTurns += 1 - if self.isEnemyInfastructure is None: - for i in ct.get_nearby_buildings(): - if ct.get_team(i) != ct.get_team(): - self.isEnemyInfastructure = ct.get_position(i) - - protectSpots = [] - for i in CardDirections: - pos = self.currentHarvester.add(i) - if self.isInBounds(ct, pos) and ct.is_in_vision(pos) and pos != self.conveyorEnd: - if ct.get_tile_env(pos) != Environment.WALL: - if ct.get_tile_building_id(pos) is None: - protectSpots.append(pos) - if len(protectSpots) == 0: - self.protectTurns = 0 - self.surroundHarvester = True - self.isEnemyInfastructure = None - self.sentinelSpot = None + if pos == tpos: + self._moveToward(ct, mi.expand(tb) & mi.passable() & ~tb) return - - if self.isEnemyInfastructure and self.sentinelSpot is None: - protectSpots.sort(key=lambda c: c.distance_squared(self.isEnemyInfastructure)) - self.sentinelSpot = protectSpots[0] - - protectSpots.sort(key=lambda c: c.distance_squared(ct.get_position())) - - for i in protectSpots: - bID = ct.get_tile_building_id(i) - etype = None - if bID is not None: - etype = ct.get_entity_type(bID) - if ct.get_position() == i and bID is None: - self.pf.moveTo(ct, self.currentHarvester) - if ct.get_position().distance_squared(i) > 2: - self.pf.moveTo(ct, i) - if i == self.sentinelSpot: - if ct.can_build_sentinel(i, i.direction_to(self.isEnemyInfastructure)): - ct.build_sentinel(i, i.direction_to(self.isEnemyInfastructure)) - return + self._moveToward(ct, mi.expand(tb) & mi.passable()) + + # ------------------------------------------------------------------ + # STATE: explore (1) + # ------------------------------------------------------------------ + def _score_explore(self, ct, mi, my_bit, others): + return 1.0, None + + def _run_explore(self, ct, mi, _): + pos = ct.get_position() + if (self.exploreTarget is not None + and manhattan(pos, self.exploreTarget) <= 2): + self.exploreTarget = None + if self.exploreTarget is None or self.stuck > 4: + frontier = mi.frontier() & mi.passable() + if self.myNum == 1: + frontier &= self._homeMask() # guard patrols home only + picked = None + if frontier: + bits = list(mi.iter_bits(frontier)) + picked = random.choice(bits[:64]) + if picked: + x, y = mi.xy(picked) + self.exploreTarget = Position(x, y) + elif self.myNum == 1 and self.teamCore is not None: + self.exploreTarget = Position( + max(0, min(mi.w - 1, + self.teamCore.x + random.randint(-4, 4))), + max(0, min(mi.h - 1, + self.teamCore.y + random.randint(-4, 4)))) else: - if ct.can_build_barrier(i): - ct.build_barrier(i) - return + self.exploreTarget = Position(random.randrange(mi.w), + random.randrange(mi.h)) + t = self.exploreTarget + self._moveToward(ct, mi.bit(t.x, t.y)) + + # ------------------------------------------------------------------ + # passive extras + # ------------------------------------------------------------------ + def _rebuildBroken(self, ct, mi): + """Replace own barriers we destroyed to walk through (Pantheon).""" + if not self.brokenBarriers or ct.get_action_cooldown() != 0: + return + pos = ct.get_position() + for p in list(self.brokenBarriers): + if pos == p: + continue # still standing on it; rebuild after stepping off + if pos.distance_squared(p) > 2: + self.brokenBarriers.remove(p) # walked away: let it go + continue + if ct.is_in_vision(p) and ct.can_build_barrier(p): + ct.build_barrier(p) + mi.own_barriers |= mi.bit(p.x, p.y) + mi.note_tile(p.x, p.y, mapinfo_mod.T_BLOCK) + self.brokenBarriers.remove(p) + return - def buildConveyor(self, ct: Controller): - ct.draw_indicator_dot(self.conveyorEnd, 255, 255, 255) - myLoc = ct.get_position() - if self.nearCore2(self.conveyorEnd) == 0: # frontier is on the core (delivery handled in buildCloser) - self.currentHarvester = None - self.conveyorEnd = None + def _tryChokeBlock(self, ct, mi): + if ct.get_action_cooldown() != 0 or not mapanalysis.chokes \ + or self.teamCore is None: return - if myLoc.distance_squared(self.conveyorEnd) > 2: - self.pf.moveTo(ct, self.conveyorEnd) - - if myLoc.distance_squared(self.conveyorEnd) < 4 : - conveyorID = ct.get_tile_building_id(self.conveyorEnd) - if conveyorID is None and self.nextTurretCountDown <= 0 and self.nearCore2(self.conveyorEnd) > 2: - if myLoc.distance_squared(self.conveyorEnd) > 0 and self.enemyPos is None and self.turretTimeOut == 0: - for i in ct.get_nearby_buildings(): # If a target exists it will find it - if ct.get_team(i) != ct.get_team(): - ct.draw_indicator_dot(ct.get_position(i), 200, 20, 100) - eType = ct.get_entity_type(i) - pos = ct.get_position(i) - if eType == EntityType.CORE: - self.enemyPos = pos - self.enemyType = 3 - elif eType in [EntityType.SENTINEL, EntityType.GUNNER] and self.enemyType < 2: - self.enemyPos = pos - self.enemyType = 2 - elif eType == EntityType.CONVEYOR and self.enemyType < 1: - self.enemyPos = pos - self.enemyType = 1 - elif eType in [EntityType.LAUNCHER, EntityType.BARRIER] and self.enemyType < 0: - self.enemyPos = pos - self.enemyType = 0 - if self.enemyPos is not None and self.turretTimeOut == 0: # If it hasn't built a turret yet it waits until it can - if ct.can_build_sentinel(self.conveyorEnd, self.conveyorEnd.direction_to(self.enemyPos)): - ct.build_sentinel(self.conveyorEnd, self.conveyorEnd.direction_to(self.enemyPos)) - self.turretTimeOut += 1 - self.nextTurretCountDown = 5 - else: + pos = ct.get_position() + enemyRef = self._enemyCoreGuess() + ti = ct.get_global_resources() + for choke in mapanalysis.chokes: + tile, clearance = choke[0], choke[1] + p = Position(tile[0], tile[1]) + if pos.distance_squared(p) > 2: + continue + if (enemyRef is not None + and manhattan(p, self.teamCore) >= manhattan(p, enemyRef)): + continue + if not self._inBounds(p) or not ct.is_in_vision(p): + continue + if ct.get_tile_building_id(p) is not None: + continue + if ct.get_tile_env(p) == Environment.ORE_TITANIUM: + continue + if clearance <= 1.2: + if ti >= ct.get_barrier_cost() + 30 and ct.can_build_barrier(p): + ct.build_barrier(p) return - if self.turretTimeOut > 0: - self.turretTimeOut += 1 - if self.enemyType == 3 and self.turretTimeOut > 0: # It's a core you're attacking, so leave the turret there - self.currentHarvester = None - self.conveyorEnd = None - self.enemyPos = None - self.enemyType = -1 - return - if self.turretTimeOut > 24: - self.enemyPos = None - self.enemyType = -1 - elif self.turretTimeOut > 0: + elif ti >= 100 and ct.can_build_launcher(p): + ct.build_launcher(p) return - self.buildCloser(ct) - def buildCloser(self, ct: Controller): - myLoc = ct.get_position() - endID = ct.get_tile_building_id(self.conveyorEnd) - if endID is not None: - endEntity = ct.get_entity_type(endID) - if endEntity in [EntityType.CONVEYOR] and ct.get_team(endID) == ct.get_team(): - self.conveyorEnd = None - self.currentHarvester = None - return - elif ct.can_destroy(self.conveyorEnd): - ct.destroy(self.conveyorEnd) - if myLoc != self.conveyorEnd: - self.pf.moveTo(ct, self.conveyorEnd) - if self.defender.destroyTurns > 5: - self.defender.placeLauncher(ct, self.conveyorEnd) - if ct.can_fire(self.conveyorEnd): - ct.fire(self.conveyorEnd) - self.defender.destroyTurns += 1 - - # Candidate cardinal conveyor extensions toward the core. - possibleConveyors = [] - for i in CardDirections: - end = self.conveyorEnd.add(i) - if self.isInBounds(ct, end): - if ct.can_build_conveyor(self.conveyorEnd, i) and (ct.get_tile_env(end) == Environment.EMPTY or ct.get_tile_env(end) == Environment.ORE_TITANIUM): - possibleConveyors.append(i) - possibleConveyors.sort(key=lambda c: self.nearCore2(self.conveyorEnd.add(c))) - - # Try to build a conveyor directly adjacent to the core - for i in possibleConveyors: - end = self.conveyorEnd.add(i) - if self.nearCore2(end) == 0: - if ct.can_build_conveyor(self.conveyorEnd, i): - ct.build_conveyor(self.conveyorEnd, i) - self.nextTurretCountDown = 0 - self.currentHarvester = None - self.conveyorEnd = None + def _healAdjacent(self, ct): + if ct.get_action_cooldown() != 0: + return + pos = ct.get_position() + if ct.get_hp() < ct.get_max_hp() and ct.can_heal(pos): + ct.heal(pos) + return + # Pantheon 6-tier heal priority: (very damaged?, type tier, damage) + TIER = {EntityType.CORE: 6, EntityType.GUNNER: 5, + EntityType.SENTINEL: 5, EntityType.LAUNCHER: 5, + EntityType.HARVESTER: 4, EntityType.CONVEYOR: 3, + EntityType.SPLITTER: 2, EntityType.BARRIER: 2} + best = None + for bid in ct.get_nearby_buildings(2): + if ct.get_team(bid) != ct.get_team(): + continue + dmg = ct.get_max_hp(bid) - ct.get_hp(bid) + if dmg >= 4: + p = ct.get_position(bid) + key = (1 if dmg >= 12 else 0, + TIER.get(ct.get_entity_type(bid), 1), dmg) + if best is None or key > best[0]: + best = (key, p) + if best is not None and ct.can_heal(best[1]): + ct.heal(best[1]) + + def _advanceAnalysis(self, ct): + job = mapanalysis.job + if job is None and ct.get_cpu_time_elapsed() < mapanalysis.CPU_BUDGET_US: + job = mapanalysis.get_job() + while (job is not None + and ct.get_cpu_time_elapsed() < mapanalysis.CPU_BUDGET_US + and job.phase != mapanalysis.DONE): + job.step() + + # ------------------------------------------------------------------ + # Turrets (ported from botv2 — proven handlers) + # ------------------------------------------------------------------ + def sentinel(self, ct: Controller) -> None: + if self.teamCore is None: + for bid in ct.get_nearby_buildings(): + if (ct.get_entity_type(bid) == EntityType.CORE + and ct.get_team(bid) == ct.get_team()): + self.teamCore = ct.get_position(bid) + team = ct.get_team() + seesEnemy = False + for eid in ct.get_nearby_entities(): + if ct.get_team(eid) != team: + seesEnemy = True + break + if seesEnemy: + self.idleTurns = 0 + else: + self.idleTurns += 1 + if self.idleTurns > 40 and ct.get_current_round() > 150: + nearHarv = any( + ct.get_team(b) == team + and ct.get_entity_type(b) == EntityType.HARVESTER + for b in ct.get_nearby_buildings(8)) + nearCore = (self.teamCore is not None + and manhattan(ct.get_position(), + self.teamCore) <= 6) + if not nearHarv and not nearCore: + ct.self_destruct() return + priority = { + EntityType.CORE: 5, EntityType.SENTINEL: 3, EntityType.GUNNER: 3, + EntityType.BARRIER: 2, EntityType.LAUNCHER: 2, + EntityType.CONVEYOR: 1, + } + targets = set(priority) + bestTile, bestKey = None, None + for tile in ct.get_attackable_tiles(): + if not self._inBounds(tile) or not ct.is_in_vision(tile): + continue + buildingID = ct.get_tile_building_id(tile) + builderID = ct.get_tile_builder_bot_id(tile) + if builderID is not None and ct.get_team(builderID) == team: + continue + if buildingID is not None and ct.get_team(buildingID) == team: + continue + tilePriority, targetHp = 0, 0 + if buildingID is not None: + btype = ct.get_entity_type(buildingID) + if btype not in targets: + continue + tilePriority = priority.get(btype, 1) + targetHp = ct.get_hp(buildingID) + if builderID is not None: + tilePriority += 2 + elif builderID is not None: + continue + key = (tilePriority, 1 if 0 < targetHp <= 18 else 0, -targetHp) + if bestTile is None or key > bestKey: + bestKey, bestTile = key, tile + if bestTile is None: + return + # Pantheon one-tap lock: if the target survives one shot and a + # LOWER-id ally sentinel also covers it, hold fire — lower id acts + # first, so next turn both shots land the same round (36 dmg kills + # anything but a core chunk) before enemy medics can heal. Bounded + # wait so a dead ally can't deadlock us. + targetHp = 0 + bid = ct.get_tile_building_id(bestTile) \ + if ct.is_in_vision(bestTile) else None + if bid is not None: + targetHp = ct.get_hp(bid) + if targetHp > 18 and self.tapWait < 2: + myId = ct.get_id() + myPos = ct.get_position() + for aid in ct.get_nearby_entities(): + if (aid < myId and ct.get_team(aid) == ct.get_team() + and ct.get_entity_type(aid) == EntityType.SENTINEL + and ct.get_position(aid).distance_squared(bestTile) + <= 32): + self.tapWait += 1 + return + self.tapWait = 0 + if ct.can_fire(bestTile): + ct.fire(bestTile) - # Try to link into an existing friendly conveyor that's closer to the core - for i in possibleConveyors: - end = self.conveyorEnd.add(i) - if self.nearCore2(end) < self.nearCore2(self.conveyorEnd): - endID = ct.get_tile_building_id(end) - if endID is not None: - endEntity = ct.get_entity_type(endID) - if endEntity == EntityType.CONVEYOR and ct.get_team(endID) == ct.get_team(): - if ct.can_build_conveyor(self.conveyorEnd, i) and ct.get_stored_resource(endID) is None: - ct.build_conveyor(self.conveyorEnd, i) - self.nextTurretCountDown = 0 - self.currentHarvester = None - self.conveyorEnd = None - self.turretTimeOut = 0 - self.enemyPos = None - self.enemyType = -1 - return - - # Extend one step closer with the best available conveyor - for i in possibleConveyors: - end = self.conveyorEnd.add(i) - endID2 = ct.get_tile_building_id(end) - if endID2 is None or ((ct.get_entity_type(endID2) == EntityType.BARRIER or ct.get_entity_type(endID2) == EntityType.LAUNCHER) and ct.get_team(endID2) == ct.get_team()): - if ct.can_build_conveyor(self.conveyorEnd, i): - ct.build_conveyor(self.conveyorEnd, i) - self.nextTurretCountDown -= 1 - self.conveyorEnd = end - self.turretTimeOut = 0 - self.enemyPos = None - self.enemyType = -1 + def gunner(self, ct: Controller) -> None: + target = ct.get_gunner_target() + if target is not None: + builder_id = ct.get_tile_builder_bot_id(target) + building_id = ct.get_tile_building_id(target) + tid = builder_id if builder_id is not None else building_id + if tid is not None and ct.get_team(tid) != ct.get_team(): + if ct.can_fire(target): + ct.fire(target) + return + if ct.get_global_resources() < 60 or ct.get_action_cooldown() != 0: + return + my_pos = ct.get_position() + current = ct.get_direction() + cands = [] + for eid in ct.get_nearby_entities(13): + et = ct.get_entity_type(eid) + if et not in (EntityType.GUNNER, EntityType.SENTINEL, + EntityType.BUILDER_BOT): + continue + if ct.get_team(eid) == ct.get_team(): + continue + ep = ct.get_position(eid) + dx, dy = ep.x - my_pos.x, ep.y - my_pos.y + if dx != 0 and dy != 0: + continue + if dx > 0: + f = Direction.EAST + elif dx < 0: + f = Direction.WEST + elif dy > 0: + f = Direction.SOUTH + elif dy < 0: + f = Direction.NORTH + else: + continue + if f == current: + continue + pr = 0 if et in (EntityType.GUNNER, EntityType.SENTINEL) else 1 + cands.append((pr, my_pos.distance_squared(ep), + CARDINALS.index(f), f)) + if cands: + f = min(cands)[3] + if ct.can_rotate(f): + ct.rotate(f) + + def launcher(self, ct: Controller) -> None: + myLoc = ct.get_position() + if self.teamCore is None: + for bid in ct.get_nearby_buildings(): + if (ct.get_entity_type(bid) == EntityType.CORE + and ct.get_team(bid) == ct.get_team()): + self.teamCore = ct.get_position(bid) + anchor = self.teamCore if self.teamCore is not None else myLoc + team = ct.get_team() + hist = self.enemyBotHistory + for uid in ct.get_nearby_units(): + if (ct.get_team(uid) != team + and ct.get_entity_type(uid) == EntityType.BUILDER_BOT): + ep = ct.get_position(uid) + lst = hist.setdefault(uid, []) + if not lst or lst[-1] != ep: + lst.append(ep) + if len(lst) > 12: + lst.pop(0) + for dx in (-1, 0, 1): + for dy in (-1, 0, 1): + if dx == 0 and dy == 0: + continue + adjacent = Position(myLoc.x + dx, myLoc.y + dy) + if (not self._inBounds(adjacent) + or not ct.is_in_vision(adjacent)): + continue + bot_id = ct.get_tile_builder_bot_id(adjacent) + if bot_id is None or ct.get_team(bot_id) == team: + continue + walked = set(hist.get(bot_id, ())[:-3]) + # Pantheon region-minimizing throw: flood the area the bot + # could navigate from each destination (enemy-POV, unseen = + # impassable) and throw into the SMALLEST pocket. Trail tiles + # (trap: forced retrace) outrank everything. + mi = self.mi + if mi is None: + w, h = ct.get_map_width(), ct.get_map_height() + self.mi = mi = mapinfo_mod.MapInfo(w, h) + mi.update_vision(ct) + epass = mi.seen & ~mi.walls & ~mi.blocked + best, bestKey = None, None + for tile in ct.get_nearby_tiles(26): + if not ct.can_launch(adjacent, tile): + continue + seed = mi.bit(tile.x, tile.y) + region = seed + for _ in range(12): + grown = mi.expand(region) & epass + if grown == region: + break + region = grown + size = bin(region).count('1') + key = (1 if tile in walked else 0, + -size, manhattan(anchor, tile)) + if bestKey is None or key > bestKey: + bestKey, best = key, tile + if best is not None: + ct.launch(adjacent, best) return - def isInBounds(self, ct: Controller, pos: Position) -> bool: - W = ct.get_map_width() - 1 - H = ct.get_map_height() - 1 - if pos.x == max(0, min(W, pos.x)) and pos.y == max(0, min(H, pos.y)): - return True - return False + # ---- misc ---- + def _inBounds(self, p): + mi = self.mi + if mi is None: + return False + return 0 <= p.x < mi.w and 0 <= p.y < mi.h + + @staticmethod + def _cardinalTo(origin, target): + dx = target.x - origin.x + dy = target.y - origin.y + if abs(dx) >= abs(dy) and dx != 0: + return Direction.EAST if dx > 0 else Direction.WEST + if dy != 0: + return Direction.SOUTH if dy > 0 else Direction.NORTH + return Direction.CENTRE diff --git a/mapanalysis.py b/mapanalysis.py new file mode 100644 index 0000000..c226589 --- /dev/null +++ b/mapanalysis.py @@ -0,0 +1,777 @@ +"""chokepoint finder for the shared tile map. + +one team-wide resumable job. builders call note_structural_tile only on +passable<->blocking changes, dupes get coalesced here. results are plain +(x, y) tuples so this file doesnt need fcode. + +smoke test, two rooms + a 1 wide corridor:: + + >>> grid = [[2] * 20 for _ in range(20)] + >>> for x in range(1, 8): + ... for y in range(3, 17): grid[x][y] = 0 + >>> for x in range(12, 19): + ... for y in range(3, 17): grid[x][y] = 0 + >>> for x in range(8, 12): grid[x][10] = 0 + >>> result, _ = analyze_for_test(grid) + >>> len(result) + 1 + >>> result[0][:2] + ((10, 10), 1) + >>> open_map = [[0] * 20 for _ in range(20)] + >>> analyze_for_test(open_map)[0] + [] + +maps are map[x][y] like Player.fullMap. +""" + +from collections import deque + + +REGION_MIN_RADIUS = 3 # min clearance for a big-room region node +MAX_CHOKE_RADIUS = 2 # widest thing we still call a choke +RATIO_SMALL = 0.7 # choke/smaller-room ratio past which rooms just merge +RATIO_LARGE = 0.5 # same but vs the bigger room +RATIO_TWO_CHOKE = 0.8 # collapse 2-exit sliver regions with a wide exit +CPU_BUDGET_US = 6000 # no analysis cpu past this point in a turn +STEP_SLICE = 200 # max queue items per step() slice + +CLEARANCE = 0 +SKELETON = 1 +PRUNE = 2 +REGIONS = 3 +CHOKES = 4 +MERGE = 5 +DONE = 6 + + +struct_version = 0 +chokes = [] +region_partition = [] +region_nodes = [] +job = None +_debug_items = [] +_debug_cursor = 0 + +_map_width = 0 +_map_height = 0 +_structural_map = [] + + +def _index(x, y, height): + return x * height + y + + +def configure_map(width, height): + """Initialise the shared structural map; unexplored tiles are blocking.""" + global _map_width, _map_height, _structural_map, struct_version, job + global chokes, region_partition, region_nodes, _debug_items, _debug_cursor + if width == _map_width and height == _map_height and _structural_map: + return + _map_width = width + _map_height = height + _structural_map = [True] * (width * height) + struct_version += 1 + job = None + chokes = [] + region_partition = [] + region_nodes = [] + _debug_items = [] + _debug_cursor = 0 + + +def reset_map(width, height): + """Start a fresh match even when its dimensions match the previous one.""" + global _map_width, _map_height, _structural_map, struct_version, job + global chokes, region_partition, region_nodes, _debug_items, _debug_cursor + _map_width = width + _map_height = height + _structural_map = [True] * (width * height) + struct_version += 1 + job = None + chokes = [] + region_partition = [] + region_nodes = [] + _debug_items = [] + _debug_cursor = 0 + + +def note_structural_tile(x, y, blocking, width, height): + """Record one structural observation and bump the version only if needed.""" + global struct_version, job, chokes, region_partition, region_nodes + global _debug_items, _debug_cursor + configure_map(width, height) + idx = _index(x, y, height) + blocking = bool(blocking) + if _structural_map[idx] == blocking: + return False + _structural_map[idx] = blocking + struct_version += 1 + job = None + chokes = [] + region_partition = [] + region_nodes = [] + _debug_items = [] + _debug_cursor = 0 + return True + + +def get_job(): + """Return the current-version singleton job, creating it lazily.""" + global job + if not _structural_map: + return None + if job is None or job.version != struct_version: + job = AnalysisJob( + _map_width, _map_height, _structural_map, struct_version + ) + return job + + +def region_at(x, y): + """Return the published region id for a tile, or -1 if none is ready.""" + if (not region_partition or x < 0 or y < 0 + or x >= _map_width or y >= _map_height): + return -1 + return region_partition[_index(x, y, _map_height)] + + +def draw_debug(ct, position_factory, deadline_us=9000): + """Draw the published red chokes and green nodes in bounded spare slices.""" + global _debug_cursor + items = _debug_items + while (_debug_cursor < len(items) + and ct.get_cpu_time_elapsed() < deadline_us): + tile, red, green, blue = items[_debug_cursor] + _debug_cursor += 1 + ct.draw_indicator_dot(position_factory(tile[0], tile[1]), + red, green, blue) + + +class AnalysisJob: + """Small-slice phase machine; callers enforce the per-turn CPU deadline.""" + + def __init__(self, width, height, blocking, version): + self.width = width + self.height = height + self.count = width * height + self.blocking = blocking[:] + self.version = version + self.phase = CLEARANCE + + self.clearance = [-1] * self.count + self.skeleton = [False] * self.count + self.adj = [[] for _ in range(self.count)] + self.active = [False] * self.count + self.degree = [0] * self.count + self.raw_region_nodes = [] + self.region_index = [-1] * self.count + self.candidates = [] + + self._mode = "seed" + self._cursor = 0 + self._queue = deque() + self._candidate = -1 + self._candidate_scan = 0 + self._candidate_end = 0 + self._candidate_greater = False + self._region_cursor = 0 + self._neighbor_cursor = 0 + self._edge_seen = set() + self._walk = None + + self.parent = [] + self.root_clearance = [] + self.root_size = [] + self.keep = [] + self.partition = [] + self._partition_clearance = [] + self._partition_size = [] + self._merge_cursor = 0 + self._ratio_buckets = [] + self._ratio_bucket = -1 + self._ratio_bucket_pos = 0 + self._incidence = [] + self._merge_changed = False + self._output = [] + + def step(self): + if self.phase == CLEARANCE: + self._step_clearance() + elif self.phase == SKELETON: + self._step_skeleton() + elif self.phase == PRUNE: + self._step_prune() + elif self.phase == REGIONS: + self._step_regions() + elif self.phase == CHOKES: + self._step_chokes() + elif self.phase == MERGE: + self._step_merge() + + def _neighbors(self, idx): + h = self.height + w = self.width + x, y = divmod(idx, h) + if x: + yield idx - h + if x + 1 < w: + yield idx + h + if y: + yield idx - 1 + if y + 1 < h: + yield idx + 1 + + def _step_clearance(self): + limit = STEP_SLICE + n = self.count + if self._mode == "seed": + end = min(n, self._cursor + limit) + for idx in range(self._cursor, end): + if self.blocking[idx]: + self.clearance[idx] = 0 + self._queue.append(idx) + self._cursor = end + if end == n: + self._mode = "bfs" if self._queue else "unbounded" + self._cursor = 0 + return + + if self._mode == "unbounded": + end = min(n, self._cursor + limit) + far = self.width + self.height + 1 + for idx in range(self._cursor, end): + if not self.blocking[idx]: + self.clearance[idx] = far + self._cursor = end + if end == n: + self.phase = SKELETON + self._mode = "mark" + self._cursor = 0 + return + + pops = 0 + clearance = self.clearance + queue = self._queue + while queue and pops < limit: + idx = queue.popleft() + next_distance = clearance[idx] + 1 + for other in self._neighbors(idx): + if clearance[other] < 0: + clearance[other] = next_distance + queue.append(other) + pops += 1 + if not queue: + self.phase = SKELETON + self._mode = "mark" + self._cursor = 0 + + def _step_skeleton(self): + n = self.count + end = min(n, self._cursor + STEP_SLICE) + if self._mode == "mark": + clearance = self.clearance + blocking = self.blocking + skeleton = self.skeleton + for idx in range(self._cursor, end): + if blocking[idx]: + continue + value = clearance[idx] + is_maximum = True + for other in self._neighbors(idx): + if clearance[other] > value: + is_maximum = False + break + skeleton[idx] = is_maximum + self._cursor = end + if end == n: + self._mode = "connect" + self._cursor = 0 + return + + skeleton = self.skeleton + adj = self.adj + h = self.height + w = self.width + for idx in range(self._cursor, end): + if not skeleton[idx]: + continue + x, y = divmod(idx, h) + if x + 1 < w and skeleton[idx + h]: + adj[idx].append(idx + h) + adj[idx + h].append(idx) + if y + 1 < h and skeleton[idx + 1]: + adj[idx].append(idx + 1) + adj[idx + 1].append(idx) + self._cursor = end + if end == n: + self.phase = PRUNE + self._mode = "init" + self._cursor = 0 + + def _step_prune(self): + if self._mode == "init": + end = min(self.count, self._cursor + STEP_SLICE) + for idx in range(self._cursor, end): + if self.skeleton[idx]: + self.active[idx] = True + degree = len(self.adj[idx]) + self.degree[idx] = degree + if degree == 1: + self._queue.append(idx) + self._cursor = end + if end == self.count: + self._mode = "peel" + return + + pops = 0 + while self._queue and pops < STEP_SLICE: + idx = self._queue.popleft() + pops += 1 + if not self.active[idx] or self.degree[idx] != 1: + continue + neighbor = -1 + for other in self.adj[idx]: + if self.active[other]: + neighbor = other + break + if neighbor < 0: + continue + if self.clearance[idx] < self.clearance[neighbor]: + self.active[idx] = False + self.degree[idx] = 0 + self.degree[neighbor] -= 1 + if self.degree[neighbor] == 1: + self._queue.append(neighbor) + if not self._queue: + self.phase = REGIONS + self._mode = "scan" + self._cursor = 0 + + def _finish_region_candidate(self): + if not self._candidate_greater: + idx = self._candidate + self.region_index[idx] = len(self.raw_region_nodes) + self.raw_region_nodes.append(idx) + self._candidate = -1 + + def _step_regions(self): + ops = 0 + h = self.height + w = self.width + while ops < STEP_SLICE: + if self._candidate >= 0: + idx = self._candidate + x, y = divmod(idx, h) + radius = self.clearance[idx] + while self._candidate_scan < self._candidate_end and ops < STEP_SLICE: + pos = self._candidate_scan + self._candidate_scan += 1 + ops += 1 + ox, oy = divmod(pos, h) + if (abs(ox - x) <= radius and abs(oy - y) <= radius + and self.active[pos] + and self.clearance[pos] > radius): + self._candidate_greater = True + self._candidate_scan = self._candidate_end + if self._candidate_scan >= self._candidate_end: + self._finish_region_candidate() + continue + + if self._cursor >= self.count: + self.phase = CHOKES + self._mode = "walk" + self._region_cursor = 0 + self._neighbor_cursor = 0 + return + + idx = self._cursor + self._cursor += 1 + ops += 1 + if not self.active[idx]: + continue + degree = self.degree[idx] + if degree != 2: + self.region_index[idx] = len(self.raw_region_nodes) + self.raw_region_nodes.append(idx) + elif self.clearance[idx] >= REGION_MIN_RADIUS: + radius = self.clearance[idx] + x, y = divmod(idx, h) + x0 = max(0, x - radius) + x1 = min(w - 1, x + radius) + y0 = max(0, y - radius) + y1 = min(h - 1, y + radius) + # scan the enclosing flat range; Chebyshev bounds are checked above + self._candidate = idx + self._candidate_scan = x0 * h + y0 + self._candidate_end = x1 * h + y1 + 1 + self._candidate_greater = False + + def _edge_key(self, left, right): + if left > right: + left, right = right, left + return left * self.count + right + + def _outside_seed(self, endpoint, inward): + """Step off a local-max plateau toward its adjacent open region.""" + best = endpoint + best_clearance = -1 + for other in self._neighbors(endpoint): + if other == inward or self.blocking[other]: + continue + value = self.clearance[other] + if value > best_clearance: + best = other + best_clearance = value + return best + + def _begin_next_walk(self): + nodes = self.raw_region_nodes + while self._region_cursor < len(nodes): + start = nodes[self._region_cursor] + neighbors = self.adj[start] + while self._neighbor_cursor < len(neighbors): + other = neighbors[self._neighbor_cursor] + self._neighbor_cursor += 1 + if not self.active[other]: + continue + key = self._edge_key(start, other) + if key in self._edge_seen: + continue + self._edge_seen.add(key) + start_region = self.region_index[start] + values = [(self.clearance[start], start), + (self.clearance[other], other)] + minimum = min(values)[0] + minima = [tile for value, tile in values if value == minimum] + self._walk = [start_region, start, other, minimum, minima, + [start, other]] + return True + self._region_cursor += 1 + self._neighbor_cursor = 0 + return False + + def _step_chokes(self): + ops = 0 + while ops < STEP_SLICE: + if self._walk is None: + if not self._begin_next_walk(): + self.phase = MERGE + self._mode = "partition_raw_init" + return + start_region, previous, current, minimum, minima, path = self._walk + ops += 1 + current_region = self.region_index[current] + if current_region >= 0 and current_region != start_region: + if minimum <= MAX_CHOKE_RADIUS: + # prefer the middle of the minimum-clearance run + chosen = minima[len(minima) // 2] + left_pos = self._outside_seed(path[0], path[1]) + right_pos = self._outside_seed(path[-1], path[-2]) + self.candidates.append( + [minimum, chosen, start_region, current_region, + left_pos, right_pos] + ) + self._walk = None + continue + + next_tile = -1 + for other in self.adj[current]: + if self.active[other] and other != previous: + next_tile = other + break + if next_tile < 0: + self._walk = None + continue + key = self._edge_key(current, next_tile) + if key in self._edge_seen: + self._walk = None + continue + self._edge_seen.add(key) + value = self.clearance[next_tile] + if value < minimum: + minimum = value + minima = [next_tile] + elif value == minimum: + minima.append(next_tile) + path.append(next_tile) + self._walk = [start_region, current, next_tile, minimum, minima, path] + + def _start_partition(self, removed): + self.partition = [-1] * self.count + self._removed = removed + self._partition_cursor = 0 + self._component = -1 + self._queue.clear() + self._partition_clearance = [] + self._partition_size = [] + + def _step_partition(self, next_mode): + ops = 0 + while ops < STEP_SLICE: + if self._queue: + idx = self._queue.popleft() + ops += 1 + self._partition_size[self._component] += 1 + value = self.clearance[idx] + if value > self._partition_clearance[self._component]: + self._partition_clearance[self._component] = value + for other in self._neighbors(idx): + if (self.partition[other] < 0 and not self.blocking[other] + and other not in self._removed): + self.partition[other] = self._component + self._queue.append(other) + continue + while self._partition_cursor < self.count: + idx = self._partition_cursor + self._partition_cursor += 1 + ops += 1 + if (not self.blocking[idx] and idx not in self._removed + and self.partition[idx] < 0): + self._component += 1 + self._partition_clearance.append(0) + self._partition_size.append(0) + self.partition[idx] = self._component + self._queue.append(idx) + break + if ops >= STEP_SLICE: + return False + if self._partition_cursor >= self.count and not self._queue: + self._mode = next_mode + return True + return False + + def _find(self, node): + parent = self.parent + while parent[node] != node: + parent[node] = parent[parent[node]] + node = parent[node] + return node + + def _union(self, left, right): + left = self._find(left) + right = self._find(right) + if left == right: + return left + lc = self.root_clearance[left] + rc = self.root_clearance[right] + if rc > lc or (rc == lc and right < left): + left, right = right, left + self.parent[right] = left + self.root_clearance[left] = max(lc, rc) + self.root_size[left] += self.root_size[right] + return left + + def _candidate_components(self, candidate): + left = self.partition[candidate[4]] + right = self.partition[candidate[5]] + if left < 0: + left = self.partition[self.raw_region_nodes[candidate[2]]] + if right < 0: + right = self.partition[self.raw_region_nodes[candidate[3]]] + return left, right + + def _step_merge(self): + if self._mode == "partition_raw_init": + self._removed = set() + self._merge_cursor = 0 + self._mode = "partition_raw_remove" + return + if self._mode == "partition_raw_remove": + end = min(len(self.candidates), self._merge_cursor + STEP_SLICE) + for idx in range(self._merge_cursor, end): + self._removed.add(self.candidates[idx][1]) + self._merge_cursor = end + if end == len(self.candidates): + self._start_partition(self._removed) + self._mode = "partition_raw" + return + if self._mode == "partition_raw": + self._step_partition("ratio_init") + return + if self._mode == "ratio_init": + count = len(self._partition_size) + self.parent = list(range(count)) + self.root_clearance = self._partition_clearance[:] + self.root_size = self._partition_size[:] + self.keep = [False] * len(self.candidates) + self._ratio_buckets = [ + [] for _ in range(MAX_CHOKE_RADIUS + 1) + ] + self._merge_cursor = 0 + self._mode = "ratio_order" + return + if self._mode == "ratio_order": + end = min(len(self.candidates), self._merge_cursor + STEP_SLICE) + for idx in range(self._merge_cursor, end): + clearance = self.candidates[idx][0] + self._ratio_buckets[clearance].append(idx) + self._merge_cursor = end + if end == len(self.candidates): + self._ratio_bucket = MAX_CHOKE_RADIUS + self._ratio_bucket_pos = 0 + self._mode = "ratio" + return + if self._mode == "ratio": + ops = 0 + while self._ratio_bucket >= 0 and ops < STEP_SLICE: + bucket = self._ratio_buckets[self._ratio_bucket] + if self._ratio_bucket_pos >= len(bucket): + self._ratio_bucket -= 1 + self._ratio_bucket_pos = 0 + continue + candidate_index = bucket[self._ratio_bucket_pos] + self._ratio_bucket_pos += 1 + ops += 1 + candidate = self.candidates[candidate_index] + left, right = self._candidate_components(candidate) + if left < 0 or right < 0: + continue + left = self._find(left) + right = self._find(right) + if left == right: + continue + choke_clearance = candidate[0] + small = min(self.root_clearance[left], self.root_clearance[right]) + large = max(self.root_clearance[left], self.root_clearance[right]) + if (choke_clearance > RATIO_SMALL * small + or choke_clearance > RATIO_LARGE * large): + self._union(left, right) + else: + self.keep[candidate_index] = True + if self._ratio_bucket < 0: + self._mode = "incidence_init" + return + if self._mode == "incidence_init": + self._incidence = [[] for _ in self.parent] + self._merge_cursor = 0 + self._mode = "incidence" + return + if self._mode == "incidence": + end = min(len(self.candidates), self._merge_cursor + STEP_SLICE) + for idx in range(self._merge_cursor, end): + if not self.keep[idx]: + continue + left, right = self._candidate_components(self.candidates[idx]) + if left < 0 or right < 0: + self.keep[idx] = False + continue + left = self._find(left) + right = self._find(right) + if left == right: + self.keep[idx] = False + continue + self._incidence[left].append(idx) + self._incidence[right].append(idx) + self._merge_cursor = end + if end == len(self.candidates): + self._merge_cursor = 0 + self._merge_changed = False + self._mode = "two_choke" + return + if self._mode == "two_choke": + end = min(len(self.parent), self._merge_cursor + STEP_SLICE) + for region in range(self._merge_cursor, end): + if self._find(region) != region: + continue + exits = self._incidence[region] + if len(exits) != 2: + continue + wider = max(exits, key=lambda i: self.candidates[i][0]) + if (self.candidates[wider][0] + > RATIO_TWO_CHOKE * self.root_clearance[region]): + left, right = self._candidate_components(self.candidates[wider]) + self._union(left, right) + self.keep[wider] = False + self._merge_changed = True + break + self._merge_cursor = end + if self._merge_changed: + self._mode = "incidence_init" + elif end == len(self.parent): + self._mode = "final_partition_init" + return + if self._mode == "final_partition_init": + self._removed = set() + self._merge_cursor = 0 + self._mode = "final_partition_remove" + return + if self._mode == "final_partition_remove": + end = min(len(self.candidates), self._merge_cursor + STEP_SLICE) + for idx in range(self._merge_cursor, end): + if self.keep[idx]: + self._removed.add(self.candidates[idx][1]) + self._merge_cursor = end + if end == len(self.candidates): + self._start_partition(self._removed) + self._mode = "final_partition" + return + if self._mode == "final_partition": + self._step_partition("publish_init") + return + if self._mode == "publish_init": + self._output = [] + self._merge_cursor = 0 + self._mode = "publish" + return + if self._mode == "publish": + self._step_publish() + return + if self._mode == "debug_publish": + self._step_debug_publish() + + def _step_publish(self): + global chokes, region_partition, region_nodes + global _debug_items, _debug_cursor + h = self.height + end = min(len(self.candidates), self._merge_cursor + STEP_SLICE) + for idx in range(self._merge_cursor, end): + candidate = self.candidates[idx] + if not self.keep[idx]: + continue + left = self.partition[candidate[4]] + right = self.partition[candidate[5]] + if left < 0 or right < 0 or left == right: + continue + tile = divmod(candidate[1], h) + self._output.append((tile, candidate[0], + self._partition_size[left], + self._partition_size[right])) + self._merge_cursor = end + if end == len(self.candidates): + chokes = self._output[:] + region_partition = self.partition[:] + region_nodes = [] + _debug_items = [(choke[0], 255, 0, 0) for choke in chokes] + _debug_cursor = 0 + self._merge_cursor = 0 + self._mode = "debug_publish" + + def _step_debug_publish(self): + global region_nodes, _debug_items + end = min(len(self.raw_region_nodes), + self._merge_cursor + STEP_SLICE) + for idx in range(self._merge_cursor, end): + node = divmod(self.raw_region_nodes[idx], self.height) + region_nodes.append(node) + _debug_items.append((node, 0, 255, 0)) + self._merge_cursor = end + if end == len(self.raw_region_nodes): + self.phase = DONE + + +def analyze_for_test(full_map): + """Synchronously analyze an ``[x][y]`` synthetic map for doctests/tests.""" + if not full_map or not full_map[0]: + return [], [] + width = len(full_map) + height = len(full_map[0]) + blocking = [] + for x in range(width): + if len(full_map[x]) != height: + raise ValueError("full_map must be rectangular and indexed [x][y]") + for y in range(height): + blocking.append(full_map[x][y] in (-1, 2, 3)) + test_job = AnalysisJob(width, height, blocking, 0) + while test_job.phase != DONE: + test_job.step() + return chokes[:], region_partition[:] diff --git a/mapinfo.py b/mapinfo.py new file mode 100644 index 0000000..66dc0e5 --- /dev/null +++ b/mapinfo.py @@ -0,0 +1,301 @@ +"""Bitmask map-info layer (Pantheon/Khaos rebuild). + +The whole map lives in Python big-int masks: bit n = x + y*W. Geometric +neighbour expansion is 4 shifts; derived masks (threat, placeable, frontier) +are boolean algebra over whole boards. On <=30x30 maps a board is <=900 bits. + +Static knowledge (seen/walls/ore) accumulates and is shared through the comm +store with the same 12-bit tile protocol botv2 uses (slots 1..6, 16 tiles per +broadcast, rotation keyed on slot 0's spawn counter). + +>>> m = MapInfo(4, 3) +>>> b = m.bit(1, 1) +>>> sorted(m.xy(t) for t in m.iter_bits(m.expand(b) & ~b)) +[(0, 1), (1, 0), (1, 2), (2, 1)] +>>> m.xy(m.east(b)) +(2, 1) +>>> m.east(m.bit(3, 1)) # east edge does not wrap +0 +""" + +from fcode import Direction, EntityType, Environment, Position + +DIR_DELTA = { + Direction.NORTH: (0, -1), Direction.SOUTH: (0, 1), + Direction.EAST: (1, 0), Direction.WEST: (-1, 0), +} + +# tile type codes shared through the store (identical to botv2) +T_EMPTY, T_ORE, T_WALL, T_BLOCK = 0, 1, 2, 3 + +GUNNER_RAY = 3 # threat approximation: gunner cross length +SENT_BAND = 5 # sentinel band length (3 wide) + + +class MapInfo: + def __init__(self, w: int, h: int): + self.w = w + self.h = h + self.n = w * h + self.board = (1 << self.n) - 1 + col0 = 0 + for y in range(h): + col0 |= 1 << (y * w) + self.not_col0 = self.board & ~col0 + self.not_colL = self.board & ~(col0 << (w - 1)) + + # accumulated static knowledge + self.seen = 0 + self.walls = 0 + self.ore = 0 + self.blocked = 0 # non-walkable buildings (shared, any team) + self.struct_version = 0 + + # remembered point features (local to this unit, refreshed on sight) + self.enemy_turrets = {} # (x,y) -> 'G' | 'S' + self.enemy_buildings = {} # (x,y) -> EntityType + self.own_harvesters = {} # (x,y) -> True + self.own_turrets = {} # (x,y) -> True (death detection) + self.turret_losses = [] # tiles where an own turret just died + self._threat_cache = (None, 0) + + # per-turn dynamic masks + self.own_bots = 0 + self.enemy_bots = 0 + self.own_conveyors = 0 + self.own_barriers = 0 # break-walkable at cost (Pantheon 15) + self.own_loaded = 0 # own conveyors holding a stack (jam-avoid) + # remembered facings of own conveyors: (x,y) -> (dx,dy) output delta + self.own_conv_facing = {} + + # outgoing share queue: list of (x, y, code) + self.out_tiles = [] + # optional hook: called (x, y, blocking) on structural changes so the + # chokepoint analyzer can invalidate incrementally + self.on_block_change = None + + # ---- geometry ---- + def bit(self, x: int, y: int) -> int: + return 1 << (x + y * self.w) + + def xy(self, single_bit: int): + n = single_bit.bit_length() - 1 + return n % self.w, n // self.w + + def east(self, m): + return (m << 1) & self.not_col0 + + def west(self, m): + return (m >> 1) & self.not_colL + + def south(self, m): + return (m << self.w) & self.board + + def north(self, m): + return m >> self.w + + def expand(self, m): + return m | self.east(m) | self.west(m) | self.south(m) | self.north(m) + + def iter_bits(self, m): + while m: + lsb = m & -m + yield lsb + m ^= lsb + + # ---- derived masks ---- + def passable(self): + """Tiles a builder could stand on (unknown treated optimistically).""" + return self.board & ~self.walls & ~self.blocked + + def threat(self): + """Soft-threat mask from remembered enemy turrets (facing unknown: + gunner = cardinal cross len 3, sentinel = 3-wide bands len 5).""" + key = tuple(sorted(self.enemy_turrets.items())) + cached, ver = self._threat_cache + if cached is not None and ver == key: + return cached + t = 0 + shifts = (self.east, self.west, self.north, self.south) + for (x, y), kind in self.enemy_turrets.items(): + b = self.bit(x, y) + if kind == 'G': + for shift in shifts: + ray = b + for _ in range(GUNNER_RAY): + ray = shift(ray) & ~self.walls + t |= ray + else: + t |= self.expand(b) + for shift in shifts: + ray = b + acc = 0 + for _ in range(SENT_BAND): + ray = shift(ray) & ~self.walls + acc |= ray + t |= self.expand(acc) # widen the ray into a 3-wide band + t &= self.board + self._threat_cache = (t, key) + return t + + def frontier(self): + """Seen tiles adjacent to unseen ones — exploration targets.""" + unseen = self.board & ~self.seen + return self.expand(unseen) & self.seen & ~self.walls + + # ---- knowledge updates ---- + def note_tile(self, x, y, code, share=True): + b = self.bit(x, y) + old_wall = self.walls & b + old_block = self.blocked & b + newly = not (self.seen & b) + self.seen |= b + if code == T_WALL: + self.walls |= b + elif code == T_BLOCK: + self.blocked |= b + else: + changed = old_block != 0 + self.blocked &= ~b + if code == T_ORE: + self.ore |= b + if changed: + self.struct_version += 1 + if self.on_block_change: + self.on_block_change(x, y, False) + if (code in (T_WALL, T_BLOCK)) and not (old_wall or old_block): + self.struct_version += 1 + if self.on_block_change: + self.on_block_change(x, y, True) + if share and (newly or (code == T_BLOCK) != bool(old_block)): + self.out_tiles.append((x, y, code)) + if len(self.out_tiles) > 96: + del self.out_tiles[:32] + + def update_vision(self, ct): + """Per-turn scan: static tiles into masks, dynamic entities rebuilt. + + Vision-delta optimization (Pantheon update_move): env is immutable, so + for already-seen tiles we skip get_tile_env entirely — walls skip ALL + queries (nothing can stand on them), other seen tiles derive env from + the ore mask and only pay the building-id query.""" + my_team = ct.get_team() + get_env = ct.get_tile_env + get_bid = ct.get_tile_building_id + note = self.note_tile + self.own_bots = 0 + self.enemy_bots = 0 + self.own_conveyors = 0 + self.own_barriers = 0 + self.own_loaded = 0 # own conveyors currently holding a stack + get_stored = ct.get_stored_resource + WALKABLE = (EntityType.CONVEYOR, EntityType.SPLITTER) + seen = self.seen + walls = self.walls + ore = self.ore + for tile in ct.get_nearby_tiles(): + x, y = tile.x, tile.y + b = 1 << (x + y * self.w) + if seen & b: + if walls & b: + continue # immutable, uninhabitable: zero queries + env = (Environment.ORE_TITANIUM if ore & b + else Environment.EMPTY) + else: + env = get_env(tile) + if env == Environment.WALL: + note(x, y, T_WALL) + continue + bid = get_bid(tile) + if bid is None: + note(x, y, T_ORE if env == Environment.ORE_TITANIUM else T_EMPTY) + self.enemy_buildings.pop((x, y), None) + self.enemy_turrets.pop((x, y), None) + self.own_harvesters.pop((x, y), None) + if self.own_turrets.pop((x, y), None): + self.turret_losses.append((x, y)) + self.own_conv_facing.pop((x, y), None) + continue + etype = ct.get_entity_type(bid) + team = ct.get_team(bid) + if etype in WALKABLE: + note(x, y, T_ORE if env == Environment.ORE_TITANIUM else T_EMPTY) + if team == my_team: + self.own_conveyors |= self.bit(x, y) + if get_stored(bid) is not None: + self.own_loaded |= self.bit(x, y) + if etype == EntityType.CONVEYOR: + d = DIR_DELTA.get(ct.get_direction(bid)) + if d is not None: + self.own_conv_facing[(x, y)] = d + else: + self.enemy_buildings[(x, y)] = etype + continue + note(x, y, T_BLOCK) + if team != my_team: + self.enemy_buildings[(x, y)] = etype + if etype == EntityType.GUNNER: + self.enemy_turrets[(x, y)] = 'G' + elif etype == EntityType.SENTINEL: + self.enemy_turrets[(x, y)] = 'S' + else: + self.enemy_buildings.pop((x, y), None) + if etype == EntityType.HARVESTER: + self.own_harvesters[(x, y)] = True + elif etype == EntityType.BARRIER: + self.own_barriers |= self.bit(x, y) + elif etype == EntityType.SENTINEL: + self.own_turrets[(x, y)] = 'S' + elif etype in (EntityType.GUNNER, EntityType.LAUNCHER): + self.own_turrets[(x, y)] = True + for uid in ct.get_nearby_units(): + if ct.get_entity_type(uid) != EntityType.BUILDER_BOT: + continue + p = ct.get_position(uid) + if ct.get_team(uid) == my_team: + self.own_bots |= self.bit(p.x, p.y) + else: + self.enemy_bots |= self.bit(p.x, p.y) + + # ---- comm store share (botv2 12-bit protocol, slots 1..6) ---- + def share(self, ct, my_num: int, core: Position): + spawn_count = ct.read_store(0) + if spawn_count <= 0 or my_num < 1 or core is None: + return + if ct.get_current_round() % spawn_count + 1 != my_num: + return + self.out_tiles.sort( + key=lambda t: abs(t[0] - core.x) + abs(t[1] - core.y), + reverse=True) + combined = 0 + for i in range(16): + if i < len(self.out_tiles): + x, y, code = self.out_tiles[i] + val = ((x & 0x1F) << 7) | ((y & 0x1F) << 2) | (code & 0x3) + else: + val = 0xFFF + combined = (combined << 12) | val + for i in range(6): + ct.write_store(6 - i, (combined >> (32 * i)) & 0xFFFFFFFF) + del self.out_tiles[:16] + + def absorb(self, ct): + combined = 0 + for i in range(1, 7): + combined = (combined << 32) | ct.read_store(i) + for i in range(16): + val = (combined >> (12 * (15 - i))) & 0xFFF + if val == 0xFFF: + continue + x = (val >> 7) & 0x1F + y = (val >> 2) & 0x1F + if x >= self.w or y >= self.h: + continue + self.note_tile(x, y, val & 0x3, share=False) + self.out_tiles = [t for t in self.out_tiles + if not (t[0] == x and t[1] == y)] + + +if __name__ == "__main__": + import doctest + doctest.testmod() diff --git a/path.py b/path.py new file mode 100644 index 0000000..7ed41d3 --- /dev/null +++ b/path.py @@ -0,0 +1,173 @@ +"""Bitmask pathfinding (Pantheon/Khaos rebuild). + +All searches operate on MapInfo big-int masks. The Dial's variant runs in +REVERSE from a target mask with a circular array of frontier bitmasks as the +bucket queue (Khaos figure 12): plain step cost 1, stepping into soft threat ++THREAT, so paths detour around enemy turret envelopes unless forced. + +dist_field() unrolls the same search into a per-tile int list for gradient +consumers (conveyor routing). +""" + +THREAT_EXTRA = 12 # entering a threat tile costs 1 + THREAT_EXTRA +BARRIER_EXTRA = 14 # entering an OWN barrier costs 1 + 14 (Pantheon 15: + # the mover destroys it, spending its action) +RING = THREAT_EXTRA + BARRIER_EXTRA + 3 # ring must exceed max edge cost + + +def next_step(mi, start_xy, targets_mask, avoid_mask=0, max_iters=None): + """Reverse Dial's from targets; returns (dx, dy) for the best first step + from start, or None if unreachable. avoid_mask tiles are impassable. + Own barriers are break-walkable at +BARRIER_EXTRA.""" + barriers = mi.own_barriers & ~avoid_mask + passable = (mi.passable() | barriers) & ~avoid_mask + targets = targets_mask & mi.board + if not targets: + return None + threat = mi.threat() + sx, sy = start_xy + sbit = mi.bit(sx, sy) + nbrs = [] + for dx, dy in ((1, 0), (-1, 0), (0, 1), (0, -1)): + nx, ny = sx + dx, sy + dy + if 0 <= nx < mi.w and 0 <= ny < mi.h: + nbrs.append((mi.bit(nx, ny), dx, dy)) + nbr_all = 0 + for b, _, _ in nbrs: + nbr_all |= b + # the start tile itself counts as reachable ground for the wavefront + passable |= sbit | targets + + ring = [0] * RING + ring[0] = targets + visited = 0 + found = {} + if max_iters is None: + max_iters = (mi.w + mi.h) * (THREAT_EXTRA + 1) + 5 + for dist in range(max_iters): + cur = ring[dist % RING] & ~visited + ring[dist % RING] = 0 + if not cur: + if not any(ring): + break + continue + visited |= cur + hit = cur & nbr_all + if hit: + for b, dx, dy in nbrs: + if (hit & b) and b not in found: + found[b] = dist + if len(found) == len(nbrs): + break + nxt = mi.expand(cur) & passable & ~visited + plain = nxt & ~threat & ~barriers + hot = nxt & threat & ~barriers + barr = nxt & barriers & ~threat + both = nxt & barriers & threat + if plain: + ring[(dist + 1) % RING] |= plain + if hot: + ring[(dist + 1 + THREAT_EXTRA) % RING] |= hot + if barr: + ring[(dist + 1 + BARRIER_EXTRA) % RING] |= barr + if both: + ring[(dist + 1 + THREAT_EXTRA + BARRIER_EXTRA) % RING] |= both + if not found: + return None + best = min(found.items(), key=lambda kv: kv[1])[0] + for b, dx, dy in nbrs: + if b == best: + return (dx, dy) + return None + + +def dist_field(mi, targets_mask, avoid_mask=0): + """Full-map cost-to-target list indexed [x + y*w]; 4096 = unreachable.""" + barriers = mi.own_barriers & ~avoid_mask + passable = (mi.passable() | barriers + | (targets_mask & mi.board)) & ~avoid_mask + threat = mi.threat() + field = [4096] * mi.n + ring = [0] * RING + ring[0] = targets_mask & mi.board + visited = 0 + max_iters = (mi.w + mi.h) * (THREAT_EXTRA + 1) + 5 + for dist in range(max_iters): + cur = ring[dist % RING] & ~visited + ring[dist % RING] = 0 + if not cur: + if not any(ring): + break + continue + visited |= cur + m = cur + while m: + lsb = m & -m + field[lsb.bit_length() - 1] = dist + m ^= lsb + nxt = mi.expand(cur) & passable & ~visited + plain = nxt & ~threat & ~barriers + hot = nxt & threat & ~barriers + barr = nxt & barriers & ~threat + both = nxt & barriers & threat + if plain: + ring[(dist + 1) % RING] |= plain + if hot: + ring[(dist + 1 + THREAT_EXTRA) % RING] |= hot + if barr: + ring[(dist + 1 + BARRIER_EXTRA) % RING] |= barr + if both: + ring[(dist + 1 + THREAT_EXTRA + BARRIER_EXTRA) % RING] |= both + return field + + +def closest(mi, start_mask, candidates_mask, avoid_mask=0, max_steps=None): + """Unweighted BFS over passable tiles; returns the candidate bit reached + first (lowest bit as tiebreak) or 0.""" + passable = mi.passable() & ~avoid_mask + front = start_mask & mi.board + visited = front + cands = candidates_mask & mi.board + if front & cands: + hit = front & cands + return hit & -hit + if max_steps is None: + max_steps = mi.w + mi.h + for _ in range(max_steps): + front = mi.expand(front) & (passable | cands) & ~visited + if not front: + return 0 + visited |= front + hit = front & cands + if hit: + return hit & -hit + return 0 + + +def claims(mi, my_bit, others_mask, candidates_mask, max_steps=None): + """Khaos claim partition: simultaneous competing floodfills. Returns the + subset of candidates MY region reaches strictly first (ties -> others, + keeping the claim conservative).""" + passable = mi.passable() + cands = candidates_mask & mi.board + if not cands: + return 0 + my_front = my_bit + other_front = others_mask & mi.board + claimed = my_front | other_front + mine = my_front & cands + cands &= ~other_front + if max_steps is None: + max_steps = mi.w + mi.h + remaining = cands & ~mine + for _ in range(max_steps): + if not remaining or not (my_front or other_front): + break + other_front = mi.expand(other_front) & (passable | cands) & ~claimed + claimed |= other_front + remaining &= ~other_front + my_front = mi.expand(my_front) & (passable | cands) & ~claimed + claimed |= my_front + mine |= my_front & remaining + remaining &= ~my_front + return mine diff --git a/pathfinding.py b/pathfinding.py deleted file mode 100644 index 9ba6647..0000000 --- a/pathfinding.py +++ /dev/null @@ -1,156 +0,0 @@ -from fcode import * - -DIRECTIONS = [ - Direction.NORTH, - Direction.EAST, - Direction.SOUTH, - Direction.WEST, -] - - -class Pathfinder: - def __init__(self): - self.minDistToTarget = 2**10 - self.objectOnRight = True - self.currentObstacle = None - self.currentTarget = None - self.canRotate = True - self.visitedStates = set() - - def reset(self): - self.currentTarget = None - self.minDistToTarget = 2**10 - self.objectOnRight = True - self.currentObstacle = None - self.canRotate = True - self.visitedStates = set() - - def isInBounds(self, c: Controller, pos: Position) -> bool: - W = c.get_map_width() - 1 - H = c.get_map_height() - 1 - return pos.x == max(0, min(W, pos.x)) and pos.y == max(0, min(H, pos.y)) - - def canMove(self, c: Controller, direction: Direction) -> bool: - dest = c.get_position().add(direction) - if not self.isInBounds(c, dest): - return False - return c.can_move(direction) - - def _move(self, c: Controller, direction: Direction): - if c.can_move(direction): - c.move(direction) - - @staticmethod - def _diagonalDist(a: Position, b: Position) -> int: - return abs(a.x - b.x) + abs(a.y - b.y) # cardinal update - - def _isKnownPassable(self, c: Controller, pos: Position) -> bool: - # Only trust this if we currently have vision on the tile -- otherwise we can't - # tell whether the remembered obstacle has actually cleared. - if not self.isInBounds(c, pos) or not c.is_in_vision(pos): - return False - if c.get_tile_env(pos) == Environment.WALL: - return False - return c.get_tile_building_id(pos) is None - - def _setInitialDirection(self, c: Controller): - myLoc = c.get_position() - forward = myLoc.direction_to(self.currentTarget) - left = forward.rotate_left() - right = forward.rotate_right() - - for _ in range(8): - if self.canMove(c, left): break - left = left.rotate_left() - for _ in range(8): - if self.canMove(c, right): break - right = right.rotate_right() - - leftLoc = myLoc.add(left) - rightLoc = myLoc.add(right) - leftDist = self._diagonalDist(leftLoc, self.currentTarget) - rightDist = self._diagonalDist(rightLoc, self.currentTarget) - - if leftDist < rightDist: - self.objectOnRight = True - elif rightDist < leftDist: - self.objectOnRight = False - else: - # Tie on Chebyshev distance -- break it with actual distance from the bot. - self.objectOnRight = myLoc.distance_squared(leftLoc) < myLoc.distance_squared(rightLoc) - - self.currentObstacle = (myLoc.add(left.rotate_right()) - if self.objectOnRight - else myLoc.add(right.rotate_left())) - - def _followWall(self, c: Controller, myLoc: Position, canRotate: bool = None): - if canRotate is None: - canRotate = self.canRotate - direction = myLoc.direction_to(self.currentObstacle) - for _ in range(8): - direction = (direction.rotate_left() if self.objectOnRight else direction.rotate_right()) - if self.canMove(c, direction): - self._move(c, direction) - return - loc = myLoc.add(direction) - if not self.isInBounds(c, loc): - if canRotate: - # Hit the map edge -- flip which side we're hugging and restart the - # sweep from scratch (rather than continuing this loop with a direction - # that points off the map), locking further flips until the next reset. - self.objectOnRight = not self.objectOnRight - self.canRotate = False - self._followWall(c, myLoc, canRotate=False) - return - continue - if not self.canMove(c, direction): - self.currentObstacle = loc - - def _state(self, c: Controller, target: Position): - myLoc = c.get_position() - ref = self.currentObstacle if self.currentObstacle is not None else target - direction = myLoc.direction_to(ref) - return (myLoc.x, myLoc.y, direction, self.objectOnRight) - - def moveTo(self, c: Controller, target: Position): - if self.currentTarget is None or self.currentTarget != target: - self.reset() - self.currentTarget = target - - if not any(self.canMove(c, d) for d in DIRECTIONS): - return - - myLoc = c.get_position() - - dist = self._diagonalDist(myLoc, target) - if dist < self.minDistToTarget: - self.reset() - self.minDistToTarget = dist - - # If the obstacle we're routing around has turned out to be passable (vision - # updated, a unit moved off it, etc.), stop treating it as blocking and retry - # a direct line instead of continuing to wall-follow around nothing. - if self.currentObstacle is not None and self._isKnownPassable(c, self.currentObstacle): - self.reset() - self.minDistToTarget = dist - - # Cycle detection: if we've already been in this exact (position, facing-toward- - # obstacle-or-target, which-side-we're-hugging) state on this trip, the wall-follow - # sweep is looping around an obstacle it can't escape -- force a reset instead of - # repeating the same loop forever. - state = self._state(c, target) - if state in self.visitedStates: - self.reset() - self.minDistToTarget = dist - else: - self.visitedStates.add(state) - - self.currentTarget = target - - if self.currentObstacle is None: - forward = myLoc.direction_to(target) - if self.canMove(c, forward): - self._move(c, forward) - return - self._setInitialDirection(c) - self._followWall(c, myLoc) diff --git a/planes.py b/planes.py new file mode 100644 index 0000000..27c280b --- /dev/null +++ b/planes.py @@ -0,0 +1,99 @@ +"""Bit-sliced score planes (Pantheon figure 14) for gunner placement. + +A plane stack is a list of NUM_PLANES big-int bitmasks; the integer score of +tile n is sum(((planes[i] >> n) & 1) << i). add_const() adds a constant to +every tile of a mask in O(NUM_PLANES) whole-board ops — cost independent of +how many tiles are in the mask. argmax_mask() extracts the best-scoring tiles +in one MSB-down pass. + +Scoring: for each facing, shift every enemy-class mask BACKWARD along the +facing k=1..3 steps with wall clipping; tiles that survive are placements +whose forward ray hits that enemy at range k. Score is discounted with range +(gunners can't shoot over buildings, so adjacency is king). +""" + +NUM_PLANES = 8 +MAX_SCORE = (1 << NUM_PLANES) - 1 + +# enemy class -> base placement value (hit at range 1) +# range discount ~0.9^k like Pantheon +RAY_STEPS = 3 +DISCOUNT = (1.0, 0.9, 0.81) + + +def add_const(planes, c, mask): + """Add constant c to every tile in mask (ripple-carry over planes).""" + if not mask or c <= 0: + return + i = 0 + while c and i < NUM_PLANES: + if c & 1: + carry = planes[i] & mask + planes[i] ^= mask + j = i + 1 + while carry and j < NUM_PLANES: + new_carry = planes[j] & carry + planes[j] ^= carry + carry = new_carry + j += 1 + c >>= 1 + i += 1 + + +def argmax_mask(planes, cands): + """(best_score, mask_of_best_tiles) among cands; (0, 0) if none.""" + if not cands: + return 0, 0 + best = 0 + cur = cands + for i in range(NUM_PLANES - 1, -1, -1): + t = cur & planes[i] + if t: + cur = t + best |= 1 << i + return best, cur + + +def gunner_planes(mi, class_masks, valid_mask): + """Per-facing plane stacks for gunner placements. + + class_masks: list of (value, enemy_mask). Returns {facing_key: planes} + with facing_key in 'E','W','N','S' — the direction the built gunner faces. + A placement facing E hits targets to its EAST, so target masks are shifted + WEST (backward) onto placement tiles. + """ + shifts_back = { + 'E': mi.west, 'W': mi.east, 'N': mi.south, 'S': mi.north, + } + out = {} + nw = ~mi.walls + for fk, back in shifts_back.items(): + planes = [0] * NUM_PLANES + for value, emask in class_masks: + if not emask: + continue + s = emask + for k in range(RAY_STEPS): + s = back(s) & nw + if not s: + break + v = int(value * DISCOUNT[k]) + if v <= 0: + continue + add_const(planes, v, s & valid_mask) + out[fk] = planes + return out + + +def best_placement(mi, class_masks, valid_mask): + """(score, x, y, facing_key) of the best gunner placement, or None.""" + stacks = gunner_planes(mi, class_masks, valid_mask) + best = None + for fk, planes in stacks.items(): + score, mask = argmax_mask(planes, valid_mask) + if score > 0 and mask: + lsb = mask & -mask + x, y = mi.xy(lsb) + if best is None or score > best[0]: + best = (score, x, y, fk) + return best diff --git a/symmetry.py b/symmetry.py new file mode 100644 index 0000000..f3001b0 --- /dev/null +++ b/symmetry.py @@ -0,0 +1,116 @@ +"""Map-symmetry prediction (Khaos blueprint Tier A, section 3). + +Florent maps are symmetric by reflection or rotation. By comparing observed +static terrain (wall / ore / empty) against its mirror under each hypothesis we +eliminate the inconsistent ones, then predict the enemy core corner from the +surviving symmetry. Default to ROTATIONAL while ambiguous (Khaos default). + +fullMap tile codes (from main.py): + -1 unseen · 0 empty/passable · 1 ore · 2 wall · 3 building/unpassable(dynamic) + +Only codes {0,1,2} are static terrain we can trust for symmetry; code 3 hides +the underlying terrain (a harvester sits on ore), so we skip those tiles. +""" + +# symmetry ids, in preference order (rotational first, Khaos default) +ROTATIONAL = 0 # 180 rotation +HORIZONTAL = 1 # flip x +VERTICAL = 2 # flip y +_ALL = (ROTATIONAL, HORIZONTAL, VERTICAL) + + +def _terrain(code: int) -> int: + """Collapse a fullMap code to a static-terrain class, or -1 if unknown.""" + # wall stays wall; ore vs empty are the two passable classes. + if code == 2: + return 2 + if code == 1: + return 1 + if code == 0: + return 0 + return -1 # -1 unseen or 3 building -> terrain unknown + + +def _mirror(x: int, y: int, w: int, h: int, sym: int): + if sym == ROTATIONAL: + return w - 1 - x, h - 1 - y + if sym == HORIZONTAL: + return w - 1 - x, y + return x, h - 1 - y # VERTICAL + + +def _core_corner(cx: int, cy: int, w: int, h: int, sym: int): + """Predicted enemy core top-left corner for a 2x2 core mirrored under sym.""" + if sym == ROTATIONAL: + return w - 2 - cx, h - 2 - cy + if sym == HORIZONTAL: + return w - 2 - cx, cy + return cx, h - 2 - cy # VERTICAL + + +class SymmetryTracker: + def __init__(self, w: int, h: int): + self.w = w + self.h = h + self.alive = list(_ALL) # surviving hypotheses + self._ver = -1 # struct_version this was computed at + self._core = None # cached (cx, cy) our core corner + + def update(self, full_map, struct_version: int, core_corner) -> None: + """Recompute survivors when the shared structural map changed. + + Cheap: <=900 tiles x 3 symmetries, only on struct_version change. + """ + if core_corner is not None: + self._core = (core_corner.x, core_corner.y) + if struct_version == self._ver and len(self.alive) <= 1: + return # already resolved, nothing new can help + if struct_version == self._ver: + return + self._ver = struct_version + w, h = self.w, self.h + alive = [] + for sym in self.alive: + ok = True + for x in range(w): + col = full_map[x] + for y in range(h): + t = _terrain(col[y]) + if t < 0: + continue + mx, my = _mirror(x, y, w, h, sym) + mt = _terrain(full_map[mx][my]) + if mt < 0: + continue + if t != mt: + ok = False + break + if not ok: + break + if ok: + alive.append(sym) + # never let noise eliminate everything; keep rotational as a floor + self.alive = alive if alive else [ROTATIONAL] + + def resolved(self) -> bool: + return len(self.alive) == 1 + + def best(self) -> int: + return self.alive[0] if self.alive else ROTATIONAL + + def enemy_core(self, core_corner): + """Best-guess enemy core corner (Position-like via factory in caller).""" + cx, cy = core_corner.x, core_corner.y + return _core_corner(cx, cy, self.w, self.h, self.best()) + + def enemy_core_candidates(self, core_corner): + """All surviving enemy-core corners, most-likely first (as (x,y)).""" + cx, cy = core_corner.x, core_corner.y + seen = set() + out = [] + for sym in self.alive: + c = _core_corner(cx, cy, self.w, self.h, sym) + if c not in seen: + seen.add(c) + out.append(c) + return out