366 lines
12 KiB
Python
366 lines
12 KiB
Python
"""Pitchwell match stub — open / build / tick / HQ victory (GAMEPLAY.md v0 slice).
|
|
|
|
Godot-less, CLI-startable. One map, two factions, one well, T1 melee path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Optional
|
|
|
|
from .catalog import (
|
|
BUILDINGS,
|
|
DROPOFF_BY_FACTION,
|
|
FACTION_HOST,
|
|
FACTION_STAFF,
|
|
HQ_BY_FACTION,
|
|
HZ,
|
|
POP_CAP_MAX,
|
|
START_PITCH,
|
|
START_POP_CAP,
|
|
T1_HALL_BY_FACTION,
|
|
T1_MELEE_BY_FACTION,
|
|
UNITS,
|
|
WORKER_BY_FACTION,
|
|
BuildSpec,
|
|
UnitSpec,
|
|
harvest_mod,
|
|
)
|
|
from .pitch_economy import SEEP_PER_SEC, SumpWell, WELL_CAPACITY
|
|
|
|
|
|
class MatchPhase(str, Enum):
|
|
OPEN = "open"
|
|
RUNNING = "running"
|
|
ENDED = "ended"
|
|
|
|
|
|
@dataclass
|
|
class Building:
|
|
spec_id: str
|
|
hp: float
|
|
max_hp: float
|
|
complete: bool = True
|
|
remaining_s: float = 0.0
|
|
well_link: Optional[int] = None # index into Match.wells when dropoff
|
|
|
|
@property
|
|
def role(self) -> str:
|
|
return BUILDINGS[self.spec_id].role
|
|
|
|
@property
|
|
def alive(self) -> bool:
|
|
return self.hp > 0.0
|
|
|
|
|
|
@dataclass
|
|
class Unit:
|
|
spec_id: str
|
|
hp: float
|
|
max_hp: float
|
|
target_hq: bool = False # v0: attack enemy HQ when True
|
|
|
|
@property
|
|
def spec(self) -> UnitSpec:
|
|
return UNITS[self.spec_id]
|
|
|
|
@property
|
|
def alive(self) -> bool:
|
|
return self.hp > 0.0
|
|
|
|
|
|
@dataclass
|
|
class QueueItem:
|
|
kind: str # building | unit
|
|
spec_id: str
|
|
remaining_s: float
|
|
well_index: Optional[int] = None # dropoff claim target
|
|
|
|
|
|
@dataclass
|
|
class Side:
|
|
faction: str
|
|
pitch: float = START_PITCH
|
|
pop_cap: int = START_POP_CAP
|
|
buildings: list[Building] = field(default_factory=list)
|
|
units: list[Unit] = field(default_factory=list)
|
|
queue: list[QueueItem] = field(default_factory=list)
|
|
claimed_wells: set[int] = field(default_factory=set)
|
|
|
|
@property
|
|
def hq(self) -> Optional[Building]:
|
|
for b in self.buildings:
|
|
if b.role == "hq" and b.alive:
|
|
return b
|
|
return None
|
|
|
|
@property
|
|
def hq_alive(self) -> bool:
|
|
return self.hq is not None
|
|
|
|
@property
|
|
def pop_used(self) -> int:
|
|
return sum(u.spec.pop for u in self.units if u.alive)
|
|
|
|
def can_afford(self, cost: float) -> bool:
|
|
return self.pitch >= cost
|
|
|
|
def has_producer(self) -> bool:
|
|
return any(b.complete and b.alive and b.role == "producer" for b in self.buildings)
|
|
|
|
def has_building(self, spec_id: str) -> bool:
|
|
return any(b.spec_id == spec_id and b.complete and b.alive for b in self.buildings)
|
|
|
|
|
|
@dataclass
|
|
class Match:
|
|
"""Two-side skirmish. Win when enemy HQ hp <= 0 (GAMEPLAY.md Modes)."""
|
|
|
|
sides: dict[str, Side]
|
|
wells: list[SumpWell] = field(default_factory=list)
|
|
phase: MatchPhase = MatchPhase.OPEN
|
|
winner: Optional[str] = None
|
|
tick_i: int = 0
|
|
sim_t: float = 0.0
|
|
hz: int = HZ
|
|
|
|
def side(self, faction: str) -> Side:
|
|
return self.sides[faction]
|
|
|
|
def enemy_of(self, faction: str) -> str:
|
|
return FACTION_HOST if faction == FACTION_STAFF else FACTION_STAFF
|
|
|
|
def start(self) -> None:
|
|
if self.phase == MatchPhase.OPEN:
|
|
self.phase = MatchPhase.RUNNING
|
|
|
|
def enqueue_building(
|
|
self, faction: str, spec_id: str, *, well_index: Optional[int] = None
|
|
) -> str:
|
|
"""Pay + queue a building. Returns ok or error token."""
|
|
if self.phase == MatchPhase.ENDED:
|
|
return "ended"
|
|
side = self.side(faction)
|
|
if spec_id not in BUILDINGS:
|
|
return "unknown_building"
|
|
spec: BuildSpec = BUILDINGS[spec_id]
|
|
if spec.role == "hq":
|
|
return "hq_not_buildable"
|
|
if not side.can_afford(spec.pitch):
|
|
return "need_pitch"
|
|
if len(side.queue) >= 8:
|
|
return "queue_full"
|
|
if spec.role == "dropoff":
|
|
if well_index is None:
|
|
return "need_well"
|
|
if well_index < 0 or well_index >= len(self.wells):
|
|
return "bad_well"
|
|
side.pitch -= spec.pitch
|
|
side.queue.append(
|
|
QueueItem("building", spec_id, spec.build_s, well_index=well_index)
|
|
)
|
|
if self.phase == MatchPhase.OPEN:
|
|
self.phase = MatchPhase.RUNNING
|
|
return "ok"
|
|
|
|
def enqueue_unit(self, faction: str, spec_id: str) -> str:
|
|
if self.phase == MatchPhase.ENDED:
|
|
return "ended"
|
|
side = self.side(faction)
|
|
if spec_id not in UNITS:
|
|
return "unknown_unit"
|
|
spec = UNITS[spec_id]
|
|
if spec.role != "worker" and not side.has_producer():
|
|
return "need_producer"
|
|
if not side.can_afford(spec.pitch):
|
|
return "need_pitch"
|
|
if side.pop_used + spec.pop > side.pop_cap:
|
|
return "need_pop"
|
|
if len(side.queue) >= 8:
|
|
return "queue_full"
|
|
side.pitch -= spec.pitch
|
|
side.queue.append(QueueItem("unit", spec_id, spec.build_s))
|
|
if self.phase == MatchPhase.OPEN:
|
|
self.phase = MatchPhase.RUNNING
|
|
return "ok"
|
|
|
|
def grant_pitch(self, faction: str, amount: float) -> None:
|
|
"""Offline cheat (GAMEPLAY.md: pitch NNNN offline-only)."""
|
|
if amount < 0:
|
|
raise ValueError("amount must be >= 0")
|
|
self.side(faction).pitch += amount
|
|
|
|
def order_attack_hq(self, faction: str) -> int:
|
|
"""All living non-worker units attack enemy HQ. Returns count ordered."""
|
|
n = 0
|
|
for u in self.side(faction).units:
|
|
if u.alive and u.spec.role != "worker":
|
|
u.target_hq = True
|
|
n += 1
|
|
return n
|
|
|
|
def _finish_building(self, side: Side, item: QueueItem) -> None:
|
|
spec = BUILDINGS[item.spec_id]
|
|
b = Building(item.spec_id, spec.hp, spec.hp, complete=True)
|
|
if spec.role == "housing":
|
|
side.pop_cap = min(POP_CAP_MAX, side.pop_cap + spec.pop_bonus)
|
|
if spec.role == "dropoff" and item.well_index is not None:
|
|
b.well_link = item.well_index
|
|
# Last finished refinery holds the seep (GAMEPLAY.md)
|
|
for other in self.sides.values():
|
|
other.claimed_wells.discard(item.well_index)
|
|
side.claimed_wells.add(item.well_index)
|
|
well = self.wells[item.well_index]
|
|
well.claim_via_refinery()
|
|
side.buildings.append(b)
|
|
|
|
def _finish_unit(self, side: Side, item: QueueItem) -> None:
|
|
spec = UNITS[item.spec_id]
|
|
side.units.append(Unit(item.spec_id, spec.hp, spec.hp))
|
|
|
|
def _advance_queue(self, side: Side, dt: float) -> None:
|
|
"""Advance serial build/train queue; carry leftover dt to the next item."""
|
|
left = dt
|
|
while side.queue and left > 0.0:
|
|
item = side.queue[0]
|
|
if item.remaining_s > left:
|
|
item.remaining_s -= left
|
|
return
|
|
left -= item.remaining_s
|
|
item.remaining_s = 0.0
|
|
if item.kind == "building":
|
|
self._finish_building(side, item)
|
|
else:
|
|
self._finish_unit(side, item)
|
|
side.queue.pop(0)
|
|
|
|
def _tick_economy(self, dt: float) -> None:
|
|
for idx, well in enumerate(self.wells):
|
|
if not well.claimed or well.remaining <= 0.0:
|
|
continue
|
|
# Owner = faction that holds claim
|
|
owner: Optional[str] = None
|
|
for fid, side in self.sides.items():
|
|
if idx in side.claimed_wells:
|
|
owner = fid
|
|
break
|
|
if owner is None:
|
|
continue
|
|
raw = well.tick_seep(dt)
|
|
self.sides[owner].pitch += raw * harvest_mod(owner)
|
|
|
|
def _tick_combat(self, dt: float) -> None:
|
|
for fid, side in self.sides.items():
|
|
enemy = self.sides[self.enemy_of(fid)]
|
|
hq = enemy.hq
|
|
if hq is None:
|
|
continue
|
|
for u in side.units:
|
|
if not u.alive or not u.target_hq:
|
|
continue
|
|
dmg = u.spec.dps * dt
|
|
hq.hp = max(0.0, hq.hp - dmg)
|
|
if hq.hp <= 0.0:
|
|
# Remove dead HQ from buildings list conceptually via alive check
|
|
break
|
|
|
|
def _check_victory(self) -> None:
|
|
alive = {fid: side.hq_alive for fid, side in self.sides.items()}
|
|
living = [fid for fid, ok in alive.items() if ok]
|
|
if len(living) == 1:
|
|
self.phase = MatchPhase.ENDED
|
|
self.winner = living[0]
|
|
elif len(living) == 0:
|
|
self.phase = MatchPhase.ENDED
|
|
self.winner = None # draw
|
|
|
|
def tick(self, dt: float) -> None:
|
|
if self.phase == MatchPhase.ENDED:
|
|
return
|
|
if self.phase == MatchPhase.OPEN:
|
|
self.phase = MatchPhase.RUNNING
|
|
for side in self.sides.values():
|
|
self._advance_queue(side, dt)
|
|
self._tick_economy(dt)
|
|
self._tick_combat(dt)
|
|
self.tick_i += 1
|
|
self.sim_t += dt
|
|
self._check_victory()
|
|
|
|
def tick_hz(self, n: int = 1) -> None:
|
|
dt = 1.0 / self.hz
|
|
for _ in range(n):
|
|
if self.phase == MatchPhase.ENDED:
|
|
break
|
|
self.tick(dt)
|
|
|
|
def snapshot(self) -> dict:
|
|
out: dict = {
|
|
"phase": self.phase.value,
|
|
"winner": self.winner,
|
|
"tick": self.tick_i,
|
|
"t": round(self.sim_t, 3),
|
|
"wells": [
|
|
{
|
|
"remaining": round(w.remaining, 2),
|
|
"claimed": w.claimed,
|
|
}
|
|
for w in self.wells
|
|
],
|
|
"sides": {},
|
|
}
|
|
for fid, side in self.sides.items():
|
|
hq = side.hq
|
|
out["sides"][fid] = {
|
|
"pitch": round(side.pitch, 2),
|
|
"pop": f"{side.pop_used}/{side.pop_cap}",
|
|
"hq_hp": None if hq is None else round(hq.hp, 1),
|
|
"buildings": [b.spec_id for b in side.buildings if b.alive],
|
|
"units": [u.spec_id for u in side.units if u.alive],
|
|
"queue": [q.spec_id for q in side.queue],
|
|
"wells": sorted(side.claimed_wells),
|
|
}
|
|
return out
|
|
|
|
|
|
def open_match(*, hz: int = HZ) -> Match:
|
|
"""Match open: each side HQ + 4 workers, 500 Pitch, one shared well (GAMEPLAY.md)."""
|
|
sides: dict[str, Side] = {}
|
|
for faction in (FACTION_STAFF, FACTION_HOST):
|
|
hq_id = HQ_BY_FACTION[faction]
|
|
hq_spec = BUILDINGS[hq_id]
|
|
worker_id = WORKER_BY_FACTION[faction]
|
|
w_spec = UNITS[worker_id]
|
|
side = Side(faction=faction)
|
|
side.buildings.append(Building(hq_id, hq_spec.hp, hq_spec.hp))
|
|
for _ in range(4):
|
|
side.units.append(Unit(worker_id, w_spec.hp, w_spec.hp))
|
|
sides[faction] = side
|
|
well = SumpWell(remaining=WELL_CAPACITY)
|
|
m = Match(sides=sides, wells=[well], hz=hz)
|
|
return m
|
|
|
|
|
|
def seed_v0_path(m: Match, faction: str) -> list[str]:
|
|
"""Enqueue Seed slice: dropoff on well 0, housing, T1 producer (GAMEPLAY.md Seed)."""
|
|
notes: list[str] = []
|
|
drop = DROPOFF_BY_FACTION[faction]
|
|
# housing / hall keyed in catalog helpers — import locally to avoid cycle noise
|
|
from .catalog import HOUSING_BY_FACTION
|
|
|
|
housing = HOUSING_BY_FACTION[faction]
|
|
hall = T1_HALL_BY_FACTION[faction]
|
|
for spec_id, kwargs in (
|
|
(drop, {"well_index": 0}),
|
|
(housing, {}),
|
|
(hall, {}),
|
|
):
|
|
r = m.enqueue_building(faction, spec_id, **kwargs)
|
|
notes.append(f"{spec_id}:{r}")
|
|
return notes
|
|
|
|
|
|
def train_t1_melee(m: Match, faction: str, n: int = 3) -> list[str]:
|
|
uid = T1_MELEE_BY_FACTION[faction]
|
|
return [f"{uid}:{m.enqueue_unit(faction, uid)}" for _ in range(n)]
|