"""Days Without Strife - Planner optimizer. This module models the *Planner* side of Days Without Strife: a Faction's Industry Actions over a number of Turns, and finds the plan that maximizes a weighted score over the end-game resources. It uses Google OR-Tools (CP-SAT). What is modeled --------------- For each Turn each controlled City may take exactly one Industry Action: * COLLECT - gain Resources based on the City's type * RENOVATE - change the City's type (not the central Metropolis) * UPGRADE - spend Steel to install one of the City's Upgrades * LAUNCH - spend 7 Steel to launch an Airship (+3 Renown) * IDLE - do nothing City types and their Collection (most cost 1 Capital): Hub : [+2 Capital] OR [spend 1 Capital -> +2 Luxuries] Foundry : 3 vats (Steel/Brass/Electrum). Collect one vat -> spend 1 Capital, gain X (the vat's current level), empty that vat, and +1 to the other two vats. Monument : spend 1 Capital -> +2 Renown to this City Metropolis : spend 1 Capital -> +2 Trade Goods and +1 Renown to this City Upgrades (Steel cost in brackets, each grants +1 City Renown unless noted): Infrastructure [0] : -1 to the cost of future Upgrades here (no Renown) Harvester [2] : +1 to Collection yield Fine Dining [2] : (Hub) +2 bonus Luxuries whenever Resources collected Overflow Vats [2] : (Foundry) the two un-collected vats get +1 extra Transit Auth. [2] : (Metropolis) +1 Express Ticket whenever collected Propaganda [2] : (Monument) military only - no resource effect Fortification [4] : +2 Renown (military defense otherwise) Governors / Overwork: The Planner Leader has *Overwork*: when appointed Governor of a City it doubles that City's Collection that Turn and waives the Capital cost, but the City cannot Collect the following Turn. The Planner can govern at most one City per Turn. Generic governor Agents may also grant a free Upgrade or bonus Trade Goods on Collect. Other modeled industrial Governor Agents (see Agent named constructors): Baron (+Trade Goods/Bastion), Builder (free type-specific Upgrade), Capitalist (+2 Capital), Vinter (+2 Luxuries), Artificer (+1 Trade Good), Metallurgist (Overflow Vats on a Foundry), Industrialist (free Infrastructure for the governed City *and every adjacent City*), Foreman (Renovate without spending the Action), Prodigy (refund <=2 Steel on Upgrade/Launch), Provisioner (+1.5 Electrum per governing Turn), Courier (one-time +3 Capital/Steel/Brass). Objective: For each scored resource n, either ``scalar_n * amount_n`` (linear) or ``scalar_n * log_mapping[n][amount_n]`` (a caller-supplied lookup table). "Renown" is scored as the total final Renown of controlled assets (Cities + Agents + Airships), each capped at 1..9. Simplifications (documented): * Trade Goods may be exchanged 1-for-1 into other resources only at the end of the horizon (for scoring), via ``tradeable_into``. * Buying Trade Goods with Electrum (2.5 each, "awful") is not modeled. * Only resource-relevant mechanics are modeled; combat/diplomacy are not. """ from __future__ import annotations import json import sys from dataclasses import dataclass, field, asdict from enum import Enum from typing import Optional from ortools.sat.python import cp_model # --------------------------------------------------------------------------- # # Constants # --------------------------------------------------------------------------- # # Stockpiled resources (Renown is handled separately - it is not stockpiled). RESOURCES = [ "capital", "luxuries", "steel", "brass", "electrum", "trade_goods", "express", ] # Integer scale applied to (possibly fractional) objective scalars / table # values so the CP-SAT objective stays integral. OBJ_SCALE = 1000 # Electrum is the one stockpiled resource that takes fractional amounts (the # Provisioner's +1.5/Turn, "awful" 2.5-each Trade Good buys, etc.). CP-SAT only # holds integers, so the Electrum pool is tracked internally in *tenths*: a # stored value of 25 means 2.5 Electrum. Every Electrum amount entering the # model is multiplied by this scale, and every Electrum amount read back out is # divided by it. All other resources stay 1:1 integers. ELECTRUM_SCALE = 10 # Upper bound used to bound resource accumulators (and AddElement domains in # log mode). Override via Problem.max_resource if your game runs hotter. DEFAULT_MAX_RESOURCE = 300 DEFAULT_MAX_VAT = 12 # foundry vats are uncapped in the rules; bounded here. RENOWN_MIN, RENOWN_MAX = 1, 9 AIRSHIP_COST_STEEL = 7 AIRSHIP_RENOWN = 3 AIRSHIP_MAX = 3 # each Faction can Launch at most 3 Airships total class CityType(str, Enum): HUB = "hub" FOUNDRY = "foundry" MONUMENT = "monument" METROPOLIS = "metropolis" class Action(str, Enum): IDLE = "idle" COLLECT = "collect" RENOVATE = "renovate" UPGRADE = "upgrade" LAUNCH = "launch" # Upgrade keys and their base Steel cost / Renown gain. UPGRADES = { "infrastructure": dict(cost=0, renown=0), "harvester": dict(cost=2, renown=1), "fine_dining": dict(cost=2, renown=1), # hub "overflow_vats": dict(cost=2, renown=1), # foundry "transit_authority": dict(cost=2, renown=1), # metropolis "propaganda": dict(cost=2, renown=1), # monument "fortification": dict(cost=4, renown=2), } # Which type-specific "third" upgrade each city type may install. TYPE_UPGRADE = { CityType.HUB: "fine_dining", CityType.FOUNDRY: "overflow_vats", CityType.METROPOLIS: "transit_authority", CityType.MONUMENT: "propaganda", } # Reverse map: which city type a type-specific upgrade belongs to. UPGRADE_OF_TYPE = {u: ct for ct, u in TYPE_UPGRADE.items()} # Upgrades that apply to any city type and persist permanently. The remaining # upgrades are type-specific (the "third" upgrade) and only exist while the City # is that type - renovating to a new type removes the old type's upgrade and # unlocks the new one. UNIVERSAL_UPGRADES = {"infrastructure", "harvester", "fortification"} # --------------------------------------------------------------------------- # # Input dataclasses # --------------------------------------------------------------------------- # @dataclass class City: name: str type: CityType # Starting city renown. Cities start at 2, except the central Metropolis at # 4 (rulebook p.18). Left as None => defaulted by type in __post_init__. renown: Optional[int] = None # Foundry vats: starting level of each vat resource. vat_steel: int = 0 vat_brass: int = 0 vat_electrum: int = 0 upgrades: list[str] = field(default_factory=list) # already-installed # Names of Cities adjacent to this one on the map. Adjacency is treated as # symmetric (declaring A adjacent to B implies B adjacent to A) and is used # by the Industrialist Governor, whose free Infrastructure Upgrade spreads # to every City adjacent to the one being governed. adjacent: list[str] = field(default_factory=list) available_turns: Optional[list[int]] = None # None => all turns can_renovate: bool = True # Metropolis cannot renovate # Hard constraint: force a specific action on a given turn (turn -> Action). forced_action: dict[int, str] = field(default_factory=dict) def __post_init__(self): if self.renown is None: self.renown = 4 if self.type == CityType.METROPOLIS else 2 def is_available(self, t: int) -> bool: return self.available_turns is None or t in self.available_turns @dataclass class Agent: """A Faction Agent that can be appointed Governor. The Planner Leader (overwork=True) is the main lever. Other industrial agents can be expressed with the generic effect flags. """ name: str overwork: bool = False # Planner: double + waive cost free_upgrade: bool = False # e.g. Brotherhood Builder bonus_trade_goods: int = 0 # e.g. Baron: +N Trade Goods on collect # --- additional industrial Governor effects (see named constructors) --- bonus_capital: int = 0 # Capitalist: +N Capital on Collect bonus_luxuries: int = 0 # Vinter: +N Luxuries on Collect grants_overflow_vats: bool = False # Metallurgist: Overflow Vats on a Foundry grants_infrastructure: bool = False # Industrialist: free Infrastructure Upgrade free_renovate: bool = False # Foreman: Renovate without spending the Action steel_refund: int = 0 # Prodigy: refund up to N Steel on Upgrade/Launch # Provisioner: +0.5*N Electrum each Turn this Agent governs a City (tracked in # half-units so the integer model can represent the 1.5 = 3 half-unit gain). governor_electrum_half: int = 0 # Courier: one-time {resource: amount} bonus the first Turn this Agent governs. onetime_governor_bonus: dict[str, int] = field(default_factory=dict) available_turns: Optional[list[int]] = None # Hard constraint: must govern this city on this turn (turn -> city name). forced_city: dict[int, str] = field(default_factory=dict) def is_available(self, t: int) -> bool: return self.available_turns is None or t in self.available_turns # -- Named constructors for known governor Agents -------------------- # @classmethod def planner(cls, name: str = "Planner", **kw) -> "Agent": """Faction Planner Leader: Overwork (double collection, waive Capital cost, lock next-Turn collect on the governed City).""" return cls(name=name, overwork=True, **kw) @classmethod def baron(cls, name: str = "Baron", bastions: int = 3, **kw) -> "Agent": """Brotherhood Baron: as Governor, gains +1 Trade Good per Bastion on the map during Collection. Defaults to 3 Bastions per request.""" return cls(name=name, bonus_trade_goods=bastions, **kw) @classmethod def builder(cls, name: str = "Builder", **kw) -> "Agent": """Brotherhood Builder: as Governor, the City gains its type-specific 3rd Upgrade for free (modeled as waiving the Steel cost of the Upgrade installed that Turn).""" return cls(name=name, free_upgrade=True, **kw) @classmethod def capitalist(cls, name: str = "Capitalist", **kw) -> "Agent": """Capitalist: as Governor, increases Capital Collection by +2.""" return cls(name=name, bonus_capital=2, **kw) @classmethod def vinter(cls, name: str = "Vinter", **kw) -> "Agent": """Vinter: as Governor, increases Luxury Collection by +2.""" return cls(name=name, bonus_luxuries=2, **kw) @classmethod def artificer(cls, name: str = "Artificer", **kw) -> "Agent": """Artificer: as Governor, Collects an extra Trade Good.""" return cls(name=name, bonus_trade_goods=1, **kw) @classmethod def metallurgist(cls, name: str = "Metallurgist", **kw) -> "Agent": """Metallurgist: as Governor of a Foundry, grants the effects of the Overflow Vats Upgrade (does not stack with the actual Upgrade).""" return cls(name=name, grants_overflow_vats=True, **kw) @classmethod def industrialist(cls, name: str = "Industrialist", **kw) -> "Agent": """Industrialist: when appointed Governor, their City and every City adjacent to it (see ``City.adjacent``) gain the Infrastructure Upgrade for free.""" return cls(name=name, grants_infrastructure=True, **kw) @classmethod def foreman(cls, name: str = "Foreman", **kw) -> "Agent": """Foreman: as Governor, Renovating their City does not prevent other Industry Actions (the City may Renovate and still take a normal Action the same Turn).""" return cls(name=name, free_renovate=True, **kw) @classmethod def prodigy(cls, name: str = "Prodigy", **kw) -> "Agent": """Prodigy: as Governor, when Upgrading or Launching an Airship, refund up to 2 Steel afterwards.""" return cls(name=name, steel_refund=2, **kw) @classmethod def provisioner(cls, name: str = "Provisioner", **kw) -> "Agent": """Provisioner: each Turn this Agent governs a City, their Faction gains 1.5 Electrum (modeled unconditionally - the "City with a Base" requirement is not modeled).""" return cls(name=name, governor_electrum_half=3, **kw) @classmethod def courier(cls, name: str = "Courier", **kw) -> "Agent": """Courier: when first appointed Governor, offers a one-time bonus of 3 Capital, 3 Steel and 3 Brass.""" return cls( name=name, onetime_governor_bonus={"capital": 3, "steel": 3, "brass": 3}, **kw, ) @dataclass class ScoreTerm: """A single additive scoring term over a resource (or "renown"). The objective is the sum of all ScoreTerms. Each term scores the amount of ``resource`` at the END of ``turn``: * linear (log_mapping is None) : ``scalar * amount`` * log (log_mapping given) : ``scalar * log_mapping[min(amount, n-1)]`` ``turn`` is 0-indexed (same convention as forced_action); ``None`` means the final Turn of the horizon. This lets e.g. steel on Turn 2 and steel on Turn 5 carry different weights, and either can use a log mapping. ``resource`` is a stockpiled resource name or "renown". Because per-Turn Renown is not tracked, a "renown" term must target the final Turn (``turn is None``). For a log term, ``log_mapping`` is indexed by the resource amount in its own internal units. That is one entry per whole unit for every resource except Electrum, which is tracked in tenths (see ELECTRUM_SCALE): an Electrum table must be sampled per tenth, so ``log_mapping[k]`` is the score at ``k / ELECTRUM_SCALE`` Electrum (the web UI builds it this way). """ resource: str scalar: float = 1.0 turn: Optional[int] = None log_mapping: Optional[list[float]] = None @dataclass class Objective: """Scoring objective, expressed as a list of additive ScoreTerms. For backward compatibility, the legacy ``mode``/``scalars``/``log_mapping`` fields are still accepted and are lowered into final-Turn ScoreTerms: * mode == "linear" : each ``scalars[key]`` -> linear term on final Turn. * mode == "log" : each ``scalars[key]`` -> log term on final Turn, using ``log_mapping[key]`` as the table. """ terms: list[ScoreTerm] = field(default_factory=list) # --- legacy convenience fields (lowered into final-Turn terms) --------- mode: str = "linear" scalars: dict[str, float] = field(default_factory=dict) log_mapping: dict[str, list[float]] = field(default_factory=dict) def all_terms(self) -> list[ScoreTerm]: """Return the explicit terms plus the lowered legacy scalars.""" out = list(self.terms) for key, s in self.scalars.items(): if not s: continue if self.mode == "linear": out.append(ScoreTerm(resource=key, scalar=s)) elif self.mode == "log": table = self.log_mapping.get(key) if table is None: raise ValueError(f"log mode requires log_mapping[{key!r}]") out.append(ScoreTerm(resource=key, scalar=s, log_mapping=table)) else: raise ValueError(f"Unknown objective mode: {self.mode}") return out @dataclass class Problem: turns: int = 5 cities: list[City] = field(default_factory=list) agents: list[Agent] = field(default_factory=list) start: dict[str, float] = field(default_factory=dict) # resource -> amount objective: Objective = field(default_factory=Objective) # Final Renown of assets that are not modeled Cities (other agents, # already-launched airships, etc.) - added as a constant to the score. extra_renown: int = 0 # Airships this Faction has already launched (counts toward the max of 3). airships_launched: int = 0 # Trade Goods may be converted 1-for-1 into these resources for scoring. tradeable_into: list[str] = field( default_factory=lambda: ["capital", "luxuries", "steel", "brass", "electrum"] ) max_resource: int = DEFAULT_MAX_RESOURCE max_vat: int = DEFAULT_MAX_VAT # Hard constraints on a resource's amount at the END of a specific Turn. # Each entry: {"turn": int, "resource": str, "op": one of >=/<=/==, "value": int}. # Turns are 0-indexed (None or omitted => final Turn). resource_constraints: list[dict] = field(default_factory=list) # Forced resource conversions on specific Turns. These are NOT decisions the # optimizer makes - each one deterministically spends ``from_amount`` of one # resource and yields ``to_amount`` of another on the given Turn. Because the # resource balance is kept non-negative, the model is forced to have the # spent resource available at that Turn. Each entry: # {"turn": int, "from": str, "to": str, "from_amount": number, "to_amount": number}. # Turns are 0-indexed (None or omitted => final Turn); ``amount`` may be given # as a shorthand for an equal ``from_amount``/``to_amount`` (1-for-1). conversions: list[dict] = field(default_factory=list) # Optional resource conversions the optimizer MAY apply on a given Turn (or # refrain from), unlike the forced ``conversions`` above. Each one is offered # up to ``max_count`` times (default 1): the model chooses an integer # 0..max_count of applications, each spending ``from_amount`` of one resource # and yielding ``to_amount`` of another on that Turn. Same entry shape as # ``conversions`` plus an optional ``max_count``; Turns are 0-indexed # (None/omitted => final Turn). optional_conversions: list[dict] = field(default_factory=list) # --------------------------------------------------------------------------- # # Output dataclasses # --------------------------------------------------------------------------- # @dataclass class CityTurnPlan: turn: int city: str action: str detail: str = "" # e.g. collect choice, upgrade name, renovate target governor: str = "" # agent appointed governor (if any) overwork: bool = False # Net resource change produced by this Action (resource -> amount). Only the # stockpiled resources this City's Action affects appear here. deltas: dict[str, float] = field(default_factory=dict) @dataclass class Solution: status: str objective_value: float | None final_resources: dict[str, float] final_renown_total: int # Resolved starting resource amounts (after rounding), so the UI can # reconstruct each Turn's opening balance for ordering moves. start_resources: dict[str, float] = field(default_factory=dict) plan: list[CityTurnPlan] = field(default_factory=list) # Mid-game Trade Goods conversions: list of {turn, resource, amount}. trade_conversions: list[dict] = field(default_factory=list) # Forced (caller-specified, non-optimizer) conversions that were applied: # list of {turn, from, to, from_amount, to_amount}. forced_conversions: list[dict] = field(default_factory=list) # Optional conversions the optimizer chose to apply (count > 0): list of # {turn, from, to, from_amount, to_amount, count}. optional_conversions: list[dict] = field(default_factory=list) # Resource amounts at the END of each turn: list of {turn, : amount}. resources_by_turn: list[dict] = field(default_factory=list) # --------------------------------------------------------------------------- # # Model construction # --------------------------------------------------------------------------- # class _Builder: def __init__(self, problem: Problem): self.p = problem self.m = cp_model.CpModel() self.T = problem.turns self.MAXR = problem.max_resource self.MAXV = problem.max_vat # resource accumulators: res[name][t] = amount at END of turn t. # res[name][-1] handled via start; we index 0..T-1 and seed turn 0. self.res: dict[str, list[cp_model.IntVar]] = {} # resource production delta per (resource, turn). self.delta: dict[tuple[str, int], list] = {} # same deltas, but attributed to the City whose Action produced them, so # each plan row (city+turn) can report its own per-resource change. Set # while a City is being built (see _cur_ci); global deltas (trade # conversion, governor one-time bonuses) are left unattributed. self.city_delta: dict[tuple[int, str, int], list] = {} self._cur_ci: Optional[int] = None # per (city_index, turn) action booleans self.act: dict[tuple[int, int, Action], cp_model.IntVar] = {} # governor assignment: gov[(agent_idx, city_idx, turn)] bool self.gov: dict[tuple[int, int, int], cp_model.IntVar] = {} # overwork[(city_idx, turn)] bool (planner governs & overworks this city) self.overwork: dict[tuple[int, int], cp_model.IntVar] = {} # final renown per city self.city_final_renown: list[cp_model.IntVar] = [] # airship launch booleans (each adds AIRSHIP_RENOWN to asset renown) self.launches: list[cp_model.IntVar] = [] # bookkeeping for solution extraction self._collect_detail: dict[tuple[int, int], dict] = {} self._upgrade_choice: dict[tuple[int, int], dict] = {} self._renovate_choice: dict[tuple[int, int], dict] = {} # Foreman: extra Renovate that does not consume the City's Action. self._foreman_renov: dict[tuple[int, int], cp_model.IntVar] = {} # -- helpers ----------------------------------------------------------- # def _mul_bool(self, expr, b: cp_model.IntVar, ub: int): """Return an IntVar equal to expr * b where b is boolean, 0<=expr<=ub.""" y = self.m.NewIntVar(0, ub, "") self.m.Add(y <= expr) self.m.Add(y <= ub * b) self.m.Add(y >= expr - ub * (1 - b)) # y >= 0 by domain return y def _add_delta(self, resource: str, t: int, expr): self.delta.setdefault((resource, t), []).append(expr) if self._cur_ci is not None: self.city_delta.setdefault((self._cur_ci, resource, t), []).append(expr) # -- main build -------------------------------------------------------- # def build(self): self._build_adjacency() self._build_actions_and_governors() self._build_city_dynamics() self._build_trade_conversion() self._build_conversions() self._build_optional_conversions() self._build_onetime_governor_bonuses() self._build_resource_balance() self._build_provisioner_electrum() self._build_renown_total() self._build_resource_constraints() self._build_objective() return self.m def _build_actions_and_governors(self): m = self.m cities, agents = self.p.cities, self.p.agents # Action selection: exactly one action per city per turn. for ci, city in enumerate(cities): for t in range(self.T): avail = city.is_available(t) vs = {} for a in Action: # Metropolis cannot renovate; renovate also needs can_renovate. if a == Action.RENOVATE and ( not city.can_renovate or city.type == CityType.METROPOLIS ): vs[a] = m.NewConstant(0) continue vs[a] = m.NewBoolVar(f"act_c{ci}_t{t}_{a.value}") self.act[(ci, t, a)] = vs[a] # store metropolis/no-renovate as constant for a in Action: self.act[(ci, t, a)] = vs[a] if not avail: # City unavailable: forced idle. for a in Action: if a != Action.IDLE: m.Add(vs[a] == 0) m.Add(vs[Action.IDLE] == 1) else: m.Add(sum(vs[a] for a in Action) == 1) # Forced action constraint. if t in city.forced_action and avail: forced = Action(city.forced_action[t]) m.Add(vs[forced] == 1) # Governor assignment. for ai, agent in enumerate(agents): for t in range(self.T): if not agent.is_available(t): continue gv = [] for ci in range(len(cities)): g = m.NewBoolVar(f"gov_a{ai}_c{ci}_t{t}") self.gov[(ai, ci, t)] = g gv.append(g) # Each agent governs at most one city per turn. m.Add(sum(gv) <= 1) # Forced governor placement. if t in agent.forced_city: target = agent.forced_city[t] ci = self._city_index(target) m.Add(self.gov[(ai, ci, t)] == 1) # Each city has at most one governor per turn. for ci in range(len(cities)): for t in range(self.T): govs = [ self.gov[(ai, ci, t)] for ai in range(len(agents)) if (ai, ci, t) in self.gov ] if govs: m.Add(sum(govs) <= 1) # Overwork bool: planner (overwork agent) governs this city. overwork_agents = [ai for ai, a in enumerate(agents) if a.overwork] for ci in range(len(cities)): for t in range(self.T): contrib = [ self.gov[(ai, ci, t)] for ai in overwork_agents if (ai, ci, t) in self.gov ] ow = m.NewBoolVar(f"overwork_c{ci}_t{t}") if contrib: m.Add(ow == sum(contrib)) # at most one planner total else: m.Add(ow == 0) self.overwork[(ci, t)] = ow # Only one overworking placement at a time is already implied by # the agent's "<=1 city" constraint. def _build_adjacency(self): """Build the symmetric City adjacency map (``self.adj[ci]`` -> set of adjacent City indices). Declaring A adjacent to B also makes B adjacent to A, so callers need only list each edge once.""" name_to_idx = {c.name: i for i, c in enumerate(self.p.cities)} self.adj: dict[int, set[int]] = {i: set() for i in range(len(self.p.cities))} for i, c in enumerate(self.p.cities): for nm in c.adjacent or []: if nm not in name_to_idx: raise KeyError( f"City {c.name!r} is adjacent to unknown City {nm!r}") j = name_to_idx[nm] if j == i: continue self.adj[i].add(j) self.adj[j].add(i) def _city_index(self, name: str) -> int: for i, c in enumerate(self.p.cities): if c.name == name: return i raise KeyError(f"Unknown city: {name}") def _build_city_dynamics(self): for ci, city in enumerate(self.p.cities): # Attribute every delta added while building this City to it. self._cur_ci = ci self._build_one_city(ci, city) self._cur_ci = None def _build_one_city(self, ci: int, city: City): m = self.m T = self.T # --- City type over time (renovation) ------------------------------ # Built first because the installability and persistence of the # type-specific (3rd) upgrade depend on the City's *current* type. # We model an active-type indicator per turn; default = starting type. # Renovation target is chosen among the 3 other resource-types. # To keep collection math tractable we only allow collection math for # the *current* type; renovating just switches future behavior. type_active = {ct: [] for ct in CityType} renov_to = {} # (ct, t) -> bool : renovate to ct on turn t for t in range(T): for ct in CityType: type_active[ct].append(m.NewBoolVar(f"type_{ci}_{ct.value}_t{t}")) m.Add(sum(type_active[ct][t] for ct in CityType) == 1) renovate_allowed = ( city.can_renovate and city.type != CityType.METROPOLIS ) for t in range(T): renov_action = self.act[(ci, t, Action.RENOVATE)] # Foreman Governor: the City may Renovate *without* spending its # Industry Action. ``foreman_renov`` is an extra Renovate that can # co-occur with a non-Renovate primary Action. foreman_src = self._gov_attr_bool( ci, t, lambda a: a.free_renovate, "foreman") if ( renovate_allowed and city.is_available(t)) else None if foreman_src is not None: foreman_renov = m.NewBoolVar(f"foremanrenov_{ci}_t{t}") m.Add(foreman_renov <= foreman_src) # Can't double up: the primary Action is already a Renovate. m.Add(renov_action + foreman_renov <= 1) else: foreman_renov = m.NewConstant(0) self._foreman_renov[(ci, t)] = foreman_renov # Combined "did the City Renovate this Turn" indicator (0/1). renov_total = renov_action + foreman_renov choices = [] for ct in CityType: if ct == CityType.METROPOLIS: continue # cannot renovate *into* metropolis r = m.NewBoolVar(f"renov_{ci}_{ct.value}_t{t}") renov_to[(ct, t)] = r choices.append(r) if renovate_allowed and choices: m.Add(sum(choices) == renov_total) else: m.Add(renov_action == 0) for ct in CityType: if (ct, t) in renov_to: m.Add(renov_to[(ct, t)] == 0) self._renovate_choice[(ci, t)] = { ct.value: renov_to[(ct, t)] for ct in CityType if (ct, t) in renov_to } # Type transition: type@t = renov target if renovated, else type@t-1. for ct in CityType: prev = ( type_active[ct][t - 1] if t > 0 else m.NewConstant(1 if ct == city.type else 0) ) r = renov_to.get((ct, t)) if r is not None: # active(ct,t) == renovated_to_ct OR (prev AND not renovating) # Linearize: active = prev - prev*renov_total + r # since renovating sets exactly one r and clears others. not_renov_keep = self._mul_bool(prev, 1 - renov_total, 1) m.Add(type_active[ct][t] == not_renov_keep + r) else: # metropolis target impossible; keep only if not renovating. not_renov_keep = self._mul_bool(prev, 1 - renov_total, 1) m.Add(type_active[ct][t] == not_renov_keep) # --- Upgrade state ------------------------------------------------- # installed[u][t] = upgrade u is installed by END of turn t. installed = {u: [] for u in UPGRADES} # Universal upgrades apply to any type; type-specific (3rd) upgrades # only exist while the City is that type. A City can install a # type-specific upgrade for every type it can ever become (its starting # type, plus the renovate targets if renovation is allowed). reachable_types = {city.type} if renovate_allowed: reachable_types |= {CityType.HUB, CityType.FOUNDRY, CityType.MONUMENT} installable = set(UNIVERSAL_UPGRADES) | { TYPE_UPGRADE[ct] for ct in reachable_types } # installs driven by the City's UPGRADE Industry Action (cost Steel) act_install = {} # (u, t) -> bool # type-specific upgrade granted for free by a Builder Governor, applied # independently of whichever Industry Action the City takes. builder_free = {} # (u, t) -> bool # any new install this turn (action or Builder); used for Renown/detail. new_install = {} # (u, t) -> bool | expr for u in UPGRADES: for t in range(T): st = m.NewBoolVar(f"inst_{ci}_{u}_t{t}") installed[u].append(st) for u in UPGRADES: preinstalled = u in city.upgrades # The type a type-specific upgrade requires (None for universal). ct_req = UPGRADE_OF_TYPE.get(u) for t in range(T): if u not in installable: # type-specific upgrade for a type this City can't become. m.Add(installed[u][t] == 0) continue # "currently the right type" gate (1 for universal upgrades). type_gate = type_active[ct_req][t] if ct_req is not None else None prev = installed[u][t - 1] if t > 0 else m.NewConstant( 1 if preinstalled else 0 ) # A type-specific upgrade is lost if the City renovates away. keep = prev if type_gate is None else self._mul_bool(prev, type_gate, 1) ai = m.NewBoolVar(f"actinst_{ci}_{u}_t{t}") act_install[(u, t)] = ai if type_gate is not None: # can only action-install while currently this type. m.Add(ai <= type_gate) ni = ai if ct_req is not None: # Builder Governor: installs the City's current type-specific # upgrade for free this turn, regardless of its action. bf_src = self._free_upgrade_bool(ci, t) if bf_src is not None: bf = m.NewBoolVar(f"builderfree_{ci}_{u}_t{t}") # bf = bf_src AND currently this type AND not already kept m.Add(bf <= bf_src) m.Add(bf <= type_gate) m.Add(bf <= 1 - keep) m.Add(bf >= bf_src + type_gate + (1 - keep) - 2) builder_free[(u, t)] = bf ni = ai + bf if u == "infrastructure": # Industrialist Governor: installs Infrastructure for free # this turn, regardless of the City's Action. The grant # spreads from the governed City to every adjacent City. inf_src = self._industrialist_infra_bool(ci, t) if inf_src is not None: bf = m.NewBoolVar(f"indinfrafree_{ci}_t{t}") # bf = inf_src AND not already installed m.Add(bf <= inf_src) m.Add(bf <= 1 - keep) m.Add(bf >= inf_src + (1 - keep) - 1) builder_free[(u, t)] = bf ni = ai + bf # installed = keep OR new ; can't (re)install if already kept. m.Add(installed[u][t] == keep + ni) m.Add(keep + ni <= 1) new_install[(u, t)] = ni # At most one *action* upgrade per turn, and only if action==UPGRADE. # (Builder-granted installs are separate and do not consume the action.) for t in range(T): ups_this_turn = [ act_install[(u, t)] for u in UPGRADES if (u, t) in act_install ] up_action = self.act[(ci, t, Action.UPGRADE)] if ups_this_turn: m.Add(sum(ups_this_turn) == up_action) else: m.Add(up_action == 0) self._upgrade_choice[(ci, t)] = { u: new_install[(u, t)] for u in UPGRADES if (u, t) in new_install } # --- Collection ---------------------------------------------------- self._build_collection(ci, city, installed, type_active) # --- Foundry vats -------------------------------------------------- self._build_vats(ci, city, installed, type_active) # --- Steel spend for upgrades & airships --------------------------- for t in range(T): # Steel cost of the upgrade installed this turn, reduced by an # already-present Infrastructure (installed by previous turn). infra_prev = installed["infrastructure"][t - 1] if t > 0 else m.NewConstant( 1 if "infrastructure" in city.upgrades else 0 ) cost_terms = [] for u in UPGRADES: # Only action-driven upgrades cost Steel; Builder-granted # installs are free. if (u, t) not in act_install: continue base = UPGRADES[u]["cost"] if base == 0: continue ni = act_install[(u, t)] # effective cost = max(0, base - infra_prev) when installing u; # reduced if infra present: base*ni - infra_prev*ni discounted = self._mul_bool(infra_prev, ni, 1) cost_terms.append(base * ni - discounted) steel_spent = sum(cost_terms) if cost_terms else 0 if cost_terms: self._add_delta("steel", t, -steel_spent) # Airship launch: -7 steel, +3 asset renown (added to renown total). launch = self.act[(ci, t, Action.LAUNCH)] self._add_delta("steel", t, -AIRSHIP_COST_STEEL * launch) self.launches.append(launch) steel_spent = steel_spent + AIRSHIP_COST_STEEL * launch # Prodigy Governor: when Upgrading or Launching an Airship, refund # up to N Steel afterwards (N = min(refund cap, Steel spent)). prodigy_agents = [a.steel_refund for a in self.p.agents if a.steel_refund] prodigy = self._gov_attr_bool( ci, t, lambda a: a.steel_refund, "prodigy") if prodigy is not None: cap = max(prodigy_agents) spent_var = m.NewIntVar(0, AIRSHIP_COST_STEEL + 8, f"steelspent_{ci}_t{t}") m.Add(spent_var == steel_spent) capped = m.NewIntVar(0, cap, f"refundcap_{ci}_t{t}") m.AddMinEquality(capped, [spent_var, m.NewConstant(cap)]) refund = self._mul_bool(capped, prodigy, cap) self._add_delta("steel", t, refund) # --- Overwork "no collect next turn" lock -------------------------- for t in range(T - 1): # if overworked at t, cannot collect at t+1 m.Add(self.act[(ci, t + 1, Action.COLLECT)] == 0).OnlyEnforceIf( self.overwork[(ci, t)] ) # Overwork requires a collection this turn (otherwise pointless, and a # governor that overworks implies the city collects). for t in range(T): m.Add(self.act[(ci, t, Action.COLLECT)] >= self.overwork[(ci, t)]) # --- Launches need available steel handled by balance; renown ------- # --- Final renown of the city -------------------------------------- self._build_city_renown(ci, city, installed, type_active) def _gov_attr_bool(self, ci: int, t: int, pred, tag: str = "gov"): """Bool that is 1 iff a Governor of City ``ci`` on Turn ``t`` satisfies ``pred(agent)``. Returns None when no such Agent could govern here. Each City has at most one Governor per Turn, so the sum is itself a boolean.""" contrib = [ self.gov[(ai, ci, t)] for ai, a in enumerate(self.p.agents) if pred(a) and (ai, ci, t) in self.gov ] if not contrib: return None b = self.m.NewBoolVar(f"{tag}_c{ci}_t{t}") self.m.Add(b == sum(contrib)) return b def _free_upgrade_bool(self, ci: int, t: int): return self._gov_attr_bool(ci, t, lambda a: a.free_upgrade, "freeup") def _industrialist_infra_bool(self, ci: int, t: int): """Bool that is 1 iff an Industrialist Governs City ``ci`` *or* any City adjacent to it on Turn ``t`` (each grants ``ci`` a free Infrastructure Upgrade). Returns None when no Industrialist could reach this City. Multiple adjacent Cities may each be Governed by an Industrialist, so the contributions are OR'd (any one is enough) rather than summed.""" sources = [ci] + sorted(self.adj[ci]) contrib = [ self.gov[(ai, cj, t)] for cj in sources for ai, a in enumerate(self.p.agents) if a.grants_infrastructure and (ai, cj, t) in self.gov ] if not contrib: return None b = self.m.NewBoolVar(f"indinfra_c{ci}_t{t}") self.m.AddMaxEquality(b, contrib) return b def _bonus_collect_expr(self, ci: int, t: int, attr: str): """Sum of ``agent.`` over Agents governing City ``ci`` on Turn ``t`` (each such Agent's per-Collection bonus). None if no contributor.""" terms = [ getattr(a, attr) * self.gov[(ai, ci, t)] for ai, a in enumerate(self.p.agents) if getattr(a, attr) and (ai, ci, t) in self.gov ] return sum(terms) if terms else None def _bonus_tg_expr(self, ci: int, t: int): return self._bonus_collect_expr(ci, t, "bonus_trade_goods") def _build_collection(self, ci, city, installed, type_active): """Add per-turn resource deltas from Collection, for each possible type.""" m = self.m T = self.T for t in range(T): collect = self.act[(ci, t, Action.COLLECT)] ow = self.overwork[(ci, t)] harv = installed["harvester"][t] # multiplier: yield doubled when overworked. We compute base gain # then add the same again when ow=1. # Capital cost of collection is 1 unless overworked (waived); # applied per-type below. # ---- HUB (capital option) : +2 capital, no capital cost -------- hub_active = type_active[CityType.HUB][t] hub_collect = self._mul_bool(collect, hub_active, 1) # Hub sub-choice: capital vs luxuries hub_lux_choice = m.NewBoolVar(f"hublux_{ci}_t{t}") m.Add(hub_lux_choice <= hub_collect) hub_cap_choice = m.NewIntVar(0, 1, f"hubcap_{ci}_t{t}") m.Add(hub_cap_choice == hub_collect - hub_lux_choice) fine = installed["fine_dining"][t] # capital option: +2 (+harvester) capital, doubled if overwork cap_base = 2 * hub_cap_choice + self._mul_bool(harv, hub_cap_choice, 1) cap_total = cap_base + self._mul_bool(cap_base, ow, self.MAXR) self._add_delta("capital", t, cap_total) # capital option fine dining bonus luxuries fd_cap = self._mul_bool(fine, hub_cap_choice, 1) fd_cap_tot = 2 * fd_cap + self._mul_bool(2 * fd_cap, ow, self.MAXR) self._add_delta("luxuries", t, fd_cap_tot) # luxuries option: spend 1 capital -> +2 (+harvester) luxuries lux_base = 2 * hub_lux_choice + self._mul_bool(harv, hub_lux_choice, 1) # fine dining adds +2 lux too fd_lux = self._mul_bool(fine, hub_lux_choice, 1) lux_base = lux_base + 2 * fd_lux lux_total = lux_base + self._mul_bool(lux_base, ow, self.MAXR) self._add_delta("luxuries", t, lux_total) # capital cost for the luxuries option (waived under overwork) lux_cost = self._mul_bool(hub_lux_choice, 1 - ow, 1) self._add_delta("capital", t, -lux_cost) # ---- MONUMENT : spend 1 capital -> +2 city renown (renown later) mon_active = type_active[CityType.MONUMENT][t] mon_collect = self._mul_bool(collect, mon_active, 1) mon_cost = self._mul_bool(mon_collect, 1 - ow, 1) self._add_delta("capital", t, -mon_cost) # renown handled in _build_city_renown via mon_collect & ow & harv # ---- METROPOLIS : spend 1 capital -> +2 TG, +1 renown ---------- met_active = type_active[CityType.METROPOLIS][t] met_collect = self._mul_bool(collect, met_active, 1) transit = installed["transit_authority"][t] met_cost = self._mul_bool(met_collect, 1 - ow, 1) self._add_delta("capital", t, -met_cost) tg_base = 2 * met_collect + self._mul_bool(harv, met_collect, 1) tg_total = tg_base + self._mul_bool(tg_base, ow, self.MAXR) self._add_delta("trade_goods", t, tg_total) # transit authority: +1 express ticket on collect exp_base = self._mul_bool(transit, met_collect, 1) exp_total = exp_base + self._mul_bool(exp_base, ow, self.MAXR) self._add_delta("express", t, exp_total) # ---- per-Collection bonuses from governor agents --------------- # Each is applied whenever the City Collects (any type) while # governed. A bonus Governor occupies the City's only Governor slot, # so these never co-occur with the Planner's Overwork doubling. # Baron -> +N Trade Goods (N = bonus per Bastion) # Artificer -> +1 Trade Good # Capitalist -> +2 Capital # Vinter -> +2 Luxuries for resource, attr in ( ("trade_goods", "bonus_trade_goods"), ("capital", "bonus_capital"), ("luxuries", "bonus_luxuries"), ): bonus = self._bonus_collect_expr(ci, t, attr) if bonus is not None: self._add_delta(resource, t, self._gate_expr(bonus, collect)) # store detail self._collect_detail[(ci, t)] = dict( hub_lux=hub_lux_choice, ) # NOTE: Foundry collection handled in _build_vats. def _gate_expr(self, expr, b): """Return expr if b else 0, for a small non-negative linear expr.""" ub = self.MAXR y = self.m.NewIntVar(0, ub, "") self.m.Add(y <= expr) self.m.Add(y <= ub * b) self.m.Add(y >= expr - ub * (1 - b)) return y def _build_vats(self, ci, city, installed, type_active): """Foundry vats and their collection. Vats grow only on collection.""" m = self.m T = self.T vat_res = {"steel": city.vat_steel, "brass": city.vat_brass, "electrum": city.vat_electrum} # vat[res][t] = level at START of turn t. vat = {r: [] for r in vat_res} for r, start in vat_res.items(): for t in range(T): v = m.NewIntVar(0, self.MAXV, f"vat_{ci}_{r}_t{t}") vat[r].append(v) m.Add(vat[r][0] == start) self._vat_choice = getattr(self, "_vat_choice", {}) for t in range(T): collect = self.act[(ci, t, Action.COLLECT)] ow = self.overwork[(ci, t)] harv = installed["harvester"][t] overflow = installed["overflow_vats"][t] # Metallurgist Governor grants the Overflow Vats effect while # governing this (Foundry) City; it does not stack with the actual # Upgrade, so the effective flag is just the OR of the two. metal = self._gov_attr_bool( ci, t, lambda a: a.grants_overflow_vats, "metal") if metal is not None: overflow_eff = m.NewBoolVar(f"overflow_eff_{ci}_t{t}") m.AddMaxEquality(overflow_eff, [overflow, metal]) overflow = overflow_eff found_active = type_active[CityType.FOUNDRY][t] found_collect = self._mul_bool(collect, found_active, 1) # choose which vat to collect (at most one, only if found_collect) # A vat may only be collected if it is non-empty: collecting an # empty vat (level 0) is disallowed. When every vat is empty the # Foundry therefore cannot Collect this Turn (found_collect == 0). pick = {} for r in vat_res: pick[r] = m.NewBoolVar(f"vatpick_{ci}_{r}_t{t}") m.Add(pick[r] <= found_collect) # nonempty[r] == 1 iff vat[r][t] >= 1. nonempty = m.NewBoolVar(f"vatnonempty_{ci}_{r}_t{t}") m.Add(vat[r][t] >= 1).OnlyEnforceIf(nonempty) m.Add(vat[r][t] == 0).OnlyEnforceIf(nonempty.Not()) m.Add(pick[r] <= nonempty) m.Add(sum(pick[r] for r in vat_res) == found_collect) self._vat_choice[(ci, t)] = pick # capital cost (waived under overwork) f_cost = self._mul_bool(found_collect, 1 - ow, 1) self._add_delta("capital", t, -f_cost) for r in vat_res: # yield if picked = vat level (+harvester), doubled if overwork picked_level = self._gate_expr(vat[r][t], pick[r]) gain_base = picked_level + self._mul_bool(harv, pick[r], 1) gain_total = gain_base + self._mul_bool(gain_base, ow, self.MAXR) # Vats hold whole units; the Electrum pool is tracked in tenths. if r == "electrum": gain_total = ELECTRUM_SCALE * gain_total self._add_delta(r, t, gain_total) # transition to next turn's vat levels if t + 1 < T: for r in vat_res: # if picked: -> 0 # else if some other vat picked (found_collect & not pick[r]): # +1 (+1 more if overflow) # else (no foundry collect): unchanged other_collect = m.NewBoolVar(f"vatother_{ci}_{r}_t{t}") # other_collect = found_collect AND not pick[r] m.Add(other_collect <= found_collect) m.Add(other_collect <= 1 - pick[r]) m.Add(other_collect >= found_collect - pick[r]) inc = other_collect + self._mul_bool(overflow, other_collect, 1) keep = self._gate_expr(vat[r][t], 1 - found_collect) # next = keep (if no foundry collect) + (vat[r][t]+inc if other) other_keep = self._gate_expr(vat[r][t], other_collect) nxt = keep + other_keep + inc # cap at MAXV m.Add(vat[r][t + 1] <= self.MAXV) m.Add(vat[r][t + 1] == nxt) def _build_city_renown(self, ci, city, installed, type_active): m = self.m T = self.T # renown accumulates from: starting renown + upgrade renown gains # + Monument collects (+2 ea, *2 if overwork) + Metropolis collects (+1) # + airship launches do NOT add to city renown (they add asset renown). gains = [] # upgrade renown: sum over upgrades newly installed of their renown. # installed[u][T-1] minus preinstalled tells if newly installed by end. for u, info in UPGRADES.items(): if info["renown"] == 0: continue end = installed[u][T - 1] pre = 1 if u in city.upgrades else 0 gains.append(info["renown"] * (end - pre)) for t in range(T): collect = self.act[(ci, t, Action.COLLECT)] mon_active = type_active[CityType.MONUMENT][t] met_active = type_active[CityType.METROPOLIS][t] mon_collect = self._mul_bool(collect, mon_active, 1) met_collect = self._mul_bool(collect, met_active, 1) # Renown is not a Resource, so Overwork (which doubles the Resources # Collected) does NOT double these Renown gains. gains.append(2 * mon_collect) # Monument: +2 Renown on collect gains.append(1 * met_collect) # Metropolis: +1 Renown on collect raw = m.NewIntVar(0, 1000, f"rawrenown_{ci}") m.Add(raw == city.renown + sum(gains)) # capped to [1, 9] capped = m.NewIntVar(RENOWN_MIN, RENOWN_MAX, f"renown_{ci}") m.AddMinEquality(capped, [raw, m.NewConstant(RENOWN_MAX)]) # raw >= 1 always (starts >=2), so min with 9 is enough; ensure >=1 self.city_final_renown.append(capped) def _build_trade_conversion(self): """Mid-game Trade Goods exchange: each Turn, convert Trade Goods 1-for-1 into any resource in ``tradeable_into``. The converted resource becomes available that same Turn (and onward). Over-conversion is prevented by the non-negative Trade Goods balance enforced each Turn.""" m = self.m self.trade_conv: dict[tuple[str, int], cp_model.IntVar] = {} # Express Tickets cannot be acquired via Trade Goods conversion. targets = [ r for r in self.p.tradeable_into if r in RESOURCES and r != "express" ] for t in range(self.T): for r in targets: c = m.NewIntVar(0, self.MAXR, f"tgconv_{r}_t{t}") self.trade_conv[(r, t)] = c # 1 Trade Good -> 1 target unit; Electrum is stored in tenths. gain = ELECTRUM_SCALE * c if r == "electrum" else c self._add_delta(r, t, gain) # +1 target resource self._add_delta("trade_goods", t, -c) # -1 Trade Good def _build_conversions(self): """Forced (non-optimizer) resource conversions on specific Turns. Each conversion deterministically spends ``from_amount`` of one resource and yields ``to_amount`` of another at the END of its Turn. They are plain constant deltas (not decisions), added before the resource balance is built, so the non-negative balance forces the model to have the spent resource available at that Turn.""" self.conversions: list[dict] = [] for c in self.p.conversions: t_in = c.get("turn") t = self.T - 1 if t_in is None else int(t_in) if not (0 <= t < self.T): raise ValueError( f"conversion turn {t_in} out of range 0..{self.T - 1}") src, dst = c["from"], c["to"] if src not in RESOURCES: raise ValueError(f"conversion 'from' unknown resource: {src!r}") if dst not in RESOURCES: raise ValueError(f"conversion 'to' unknown resource: {dst!r}") amount = c.get("amount") raw_from = c.get("from_amount", amount if amount is not None else 0) raw_to = c.get("to_amount", amount if amount is not None else raw_from) # Electrum is tracked in tenths, so its amounts keep one decimal of # precision; every other resource stays an integer. from_scale = ELECTRUM_SCALE if src == "electrum" else 1 to_scale = ELECTRUM_SCALE if dst == "electrum" else 1 from_amt = int(round(raw_from * from_scale)) to_amt = int(round(raw_to * to_scale)) if from_amt < 0 or to_amt < 0: raise ValueError("conversion amounts must be non-negative") self._add_delta(src, t, -from_amt) self._add_delta(dst, t, to_amt) self.conversions.append({ "turn": t, "from": src, "to": dst, "from_amount": from_amt / from_scale if from_scale != 1 else from_amt, "to_amount": to_amt / to_scale if to_scale != 1 else to_amt, }) def _build_optional_conversions(self): """Optional (optimizer-chosen) resource conversions on specific Turns. Unlike the forced ``conversions``, each of these is a *decision*: the model picks an integer count in ``0..max_count`` of how many times to apply it on its Turn, spending ``from_amount`` per application of one resource and yielding ``to_amount`` of another. A count of 0 means the conversion is simply skipped. The non-negative resource balance still forces the spent resource to be available whenever count > 0.""" m = self.m # Parallel to self.conversions: meta dict + the chosen-count IntVar. self.optional_conversions: list[dict] = [] self._opt_conv_vars: list[tuple[dict, cp_model.IntVar]] = [] for i, c in enumerate(self.p.optional_conversions): t_in = c.get("turn") t = self.T - 1 if t_in is None else int(t_in) if not (0 <= t < self.T): raise ValueError( f"optional conversion turn {t_in} out of range 0..{self.T - 1}") src, dst = c["from"], c["to"] if src not in RESOURCES: raise ValueError(f"optional conversion 'from' unknown resource: {src!r}") if dst not in RESOURCES: raise ValueError(f"optional conversion 'to' unknown resource: {dst!r}") amount = c.get("amount") raw_from = c.get("from_amount", amount if amount is not None else 0) raw_to = c.get("to_amount", amount if amount is not None else raw_from) # Electrum is tracked in tenths, so its amounts keep one decimal of # precision; every other resource stays an integer. from_scale = ELECTRUM_SCALE if src == "electrum" else 1 to_scale = ELECTRUM_SCALE if dst == "electrum" else 1 from_amt = int(round(raw_from * from_scale)) to_amt = int(round(raw_to * to_scale)) if from_amt < 0 or to_amt < 0: raise ValueError("optional conversion amounts must be non-negative") max_count = int(c.get("max_count", 1)) if max_count < 1: raise ValueError("optional conversion max_count must be >= 1") n = m.NewIntVar(0, max_count, f"optconv{i}_t{t}") self._add_delta(src, t, -from_amt * n) self._add_delta(dst, t, to_amt * n) meta = { "turn": t, "from": src, "to": dst, "from_amount": from_amt / from_scale if from_scale != 1 else from_amt, "to_amount": to_amt / to_scale if to_scale != 1 else to_amt, "max_count": max_count, } self.optional_conversions.append(meta) self._opt_conv_vars.append((meta, n)) def _build_onetime_governor_bonuses(self): """Courier: the first Turn the Agent governs any City, grant a one-time ``{resource: amount}`` bonus. Added to the resource deltas before the balance is built.""" m = self.m T = self.T ncities = len(self.p.cities) for ai, a in enumerate(self.p.agents): if not a.onetime_governor_bonus: continue # gov_any[t]: this Agent governs some City on Turn t (0/1, since an # Agent governs at most one City per Turn). gov_any = [] for t in range(T): terms = [self.gov[(ai, ci, t)] for ci in range(ncities) if (ai, ci, t) in self.gov] ga = m.NewBoolVar(f"govany_a{ai}_t{t}") m.Add(ga == (sum(terms) if terms else 0)) gov_any.append(ga) for t in range(T): first = m.NewBoolVar(f"firstgov_a{ai}_t{t}") if t == 0: m.Add(first == gov_any[0]) else: ever_before = m.NewBoolVar(f"evergov_a{ai}_t{t}") m.AddMaxEquality(ever_before, gov_any[:t]) m.Add(first <= gov_any[t]) m.Add(first <= 1 - ever_before) m.Add(first >= gov_any[t] - ever_before) for r, amt in a.onetime_governor_bonus.items(): if r in RESOURCES and amt: scaled_amt = amt * ELECTRUM_SCALE if r == "electrum" else amt self._add_delta(r, t, scaled_amt * first) def _build_resource_balance(self): m = self.m T = self.T for r in RESOURCES: scale = ELECTRUM_SCALE if r == "electrum" else 1 start = int(round(self.p.start.get(r, 0) * scale)) self.res[r] = [] for t in range(T): v = m.NewIntVar(0, self.MAXR * scale, f"res_{r}_t{t}") self.res[r].append(v) prev = self.res[r][t - 1] if t > 0 else start deltas = self.delta.get((r, t), []) m.Add(v == prev + sum(deltas)) # resources must never go negative (enforced by domain >= 0). def _build_provisioner_electrum(self): """Provisioner: +1.5 Electrum each Turn it governs a City. Electrum is never spent in this model (only accumulated for scoring), so the bonus is layered on top of the Electrum pool as an ``electrum_eff[t]`` amount used by scoring and constraints. Both the pool and this overlay are in tenths (see ELECTRUM_SCALE), so the 1.5 gain is an exact 15-tenths integer with no rounding. ``governor_ electrum_half`` is in half-units (1 half = 0.5 = 5 tenths).""" m = self.m T = self.T ncities = len(self.p.cities) # default: no Provisioner effect, effective Electrum == the pool (tenths). self.electrum_eff = list(self.res["electrum"]) tenth_terms = {t: [] for t in range(T)} any_tenths = False for ai, a in enumerate(self.p.agents): if not a.governor_electrum_half: continue for ci in range(ncities): for t in range(T): if (ai, ci, t) in self.gov: tenth_terms[t].append( a.governor_electrum_half * 5 * self.gov[(ai, ci, t)]) any_tenths = True if not any_tenths: return per_turn_max = sum(a.governor_electrum_half for a in self.p.agents) * 5 bound = per_turn_max * T eff = [] cum_prev = 0 for t in range(T): cum = m.NewIntVar(0, bound, f"electenth_cum_t{t}") m.Add(cum == cum_prev + sum(tenth_terms[t])) cum_prev = cum e = m.NewIntVar(0, self.MAXR * ELECTRUM_SCALE + bound, f"electrum_eff_t{t}") m.Add(e == self.res["electrum"][t] + cum) eff.append(e) self.electrum_eff = eff _OPS = {">=", "<=", "=="} def _resource_at(self, resource: str, turn: Optional[int]): """Resource amount var at END of ``turn`` (None => final Turn). ``resource`` may be "renown" only for the final Turn, since per-Turn Renown is not tracked.""" T = self.T t = T - 1 if turn is None else turn if not (0 <= t < T): raise ValueError(f"turn {turn} out of range 0..{T - 1}") if resource == "renown": if turn is not None and turn != T - 1: raise ValueError("'renown' is only available on the final Turn") return self.renown_total if resource not in RESOURCES: raise ValueError(f"Unknown resource: {resource!r}") if resource == "electrum": # Effective Electrum includes the Provisioner Governor bonus. return self.electrum_eff[t] return self.res[resource][t] def _build_renown_total(self): m = self.m # Each Faction can Launch at most 3 Airships total (incl. any already # launched before the planning horizon). if self.launches: m.Add(sum(self.launches) <= AIRSHIP_MAX - self.p.airships_launched) # Renown total = sum of city final renown + extra + launched airships. renown_total = m.NewIntVar(0, 100000, "renown_total") launch_renown = AIRSHIP_RENOWN * sum(self.launches) if self.launches else 0 m.Add(renown_total == sum(self.city_final_renown) + self.p.extra_renown + launch_renown) self.renown_total = renown_total def _build_resource_constraints(self): m = self.m for c in self.p.resource_constraints: op = c.get("op", ">=") if op not in self._OPS: raise ValueError(f"resource_constraints op must be one of {self._OPS}") var = self._resource_at(c["resource"], c.get("turn")) # Electrum is compared in tenths; its bound may carry a decimal. scale = ELECTRUM_SCALE if c["resource"] == "electrum" else 1 value = int(round(float(c["value"]) * scale)) if op == ">=": m.Add(var >= value) elif op == "<=": m.Add(var <= value) else: m.Add(var == value) def _build_objective(self): m = self.m T = self.T obj = self.p.objective terms = [] # Trade Goods conversion is handled mid-game (see _build_trade_conversion), # so end-of-game amounts are simply the resource balances at the last Turn. # Electrum uses its effective amount (incl. the Provisioner bonus). self.final_amt = {r: self._resource_at(r, None) for r in RESOURCES} def scaled(x: float) -> int: return int(round(x * OBJ_SCALE)) # Every scoring term is a (turn, resource, scalar, log_mapping) tuple; # legacy mode/scalars are lowered into final-Turn terms by all_terms(). for term in obj.all_terms(): if not term.scalar: continue amt_var = self._resource_at(term.resource, term.turn) # Electrum's amount var is in tenths; the table/scalar are stated per # whole Electrum, so account for the scale on the way into the score. is_elec = term.resource == "electrum" if term.log_mapping is None: if is_elec: # scalar * (amt/ELECTRUM_SCALE), kept integral via OBJ_SCALE. coeff = int(round(term.scalar * OBJ_SCALE / ELECTRUM_SCALE)) terms.append(coeff * amt_var) else: terms.append(scaled(term.scalar) * amt_var) else: table = term.log_mapping # value = table[min(amt, len-1)]; scale table values to ints. # The table is indexed in the resource's own internal units, so # ``amt_var`` indexes it directly: whole units for most resources, # but tenths for Electrum (see ELECTRUM_SCALE) -- callers must # therefore supply an Electrum table sampled per tenth (entry k is # the score at k/ELECTRUM_SCALE Electrum), which the UI does. vals = [scaled(v) for v in table] tag = f"{term.resource}_t{term.turn if term.turn is not None else T - 1}" idx = m.NewIntVar(0, len(table) - 1, f"idx_{tag}") m.AddMinEquality(idx, [amt_var, m.NewConstant(len(table) - 1)]) val = m.NewIntVar(min(vals), max(vals), f"logval_{tag}") m.AddElement(idx, vals, val) terms.append(int(round(term.scalar)) * val) m.Maximize(sum(terms)) # --------------------------------------------------------------------------- # # Solve entry point # --------------------------------------------------------------------------- # def solve(problem: Problem, max_time_seconds: float = 30.0, workers: int = 8, solver_sink=None) -> Solution: builder = _Builder(problem) model = builder.build() solver = cp_model.CpSolver() solver.parameters.max_time_in_seconds = max_time_seconds solver.parameters.num_search_workers = workers # Hand the solver to the caller (if any) so it can call StopSearch() to # abort the search early, e.g. when the requesting client disconnects. if solver_sink is not None: solver_sink(solver) status = solver.Solve(model) status_name = solver.StatusName(status) if status not in (cp_model.OPTIMAL, cp_model.FEASIBLE): return Solution( status=status_name, objective_value=None, final_resources={}, final_renown_total=0, plan=[], ) return _extract(problem, builder, solver, status_name) def _extract(problem: Problem, b: _Builder, solver, status_name: str) -> Solution: T = b.T # Resources an Agent grants as a side-effect of *governing* (not produced by # the City's own Industry Action, so not in city_delta): the Provisioner's # +1.5 Electrum per governing Turn and the Courier's one-time bonus. These # are attributed to the governed City's row for that Turn. extra: dict[tuple[int, int], dict[str, float]] = {} def _add_extra(ci, t, res, amt): if amt: d = extra.setdefault((ci, t), {}) d[res] = d.get(res, 0.0) + amt for ai, agent in enumerate(problem.agents): if agent.governor_electrum_half: for ci in range(len(problem.cities)): for t in range(T): g = b.gov.get((ai, ci, t)) if g is not None and solver.Value(g) == 1: _add_extra(ci, t, "electrum", agent.governor_electrum_half / 2.0) if agent.onetime_governor_bonus: # Only the first Turn this Agent governs any City pays the bonus. first = next( ((ci, t) for t in range(T) for ci in range(len(problem.cities)) if (g := b.gov.get((ai, ci, t))) is not None and solver.Value(g) == 1), None) if first is not None: ci, t = first for res, amt in agent.onetime_governor_bonus.items(): if res in RESOURCES: _add_extra(ci, t, res, float(amt)) plan: list[CityTurnPlan] = [] for ci, city in enumerate(problem.cities): for t in range(T): chosen = None for a in Action: v = b.act[(ci, t, a)] if solver.Value(v) == 1: chosen = a break # Foreman may add a Renovate that does not consume the Action. foreman_renov = ( (ci, t) in b._foreman_renov and bool(solver.Value(b._foreman_renov[(ci, t)])) ) if (chosen is None or chosen == Action.IDLE) and not foreman_renov \ and (ci, t) not in extra: continue detail = "" governor = "" overwork = bool(solver.Value(b.overwork[(ci, t)])) # governor name for ai, agent in enumerate(problem.agents): if (ai, ci, t) in b.gov and solver.Value(b.gov[(ai, ci, t)]) == 1: governor = agent.name if chosen == Action.COLLECT: # determine collect detail pick = b._vat_choice.get((ci, t)) if pick and any(solver.Value(pick[r]) for r in pick): r = next(r for r in pick if solver.Value(pick[r])) detail = f"foundry vat: {r}" else: cd = b._collect_detail.get((ci, t), {}) if "hub_lux" in cd and solver.Value(cd["hub_lux"]) == 1: detail = "hub: +2 luxuries" else: detail = "collect" elif chosen == Action.UPGRADE: uc = b._upgrade_choice.get((ci, t), {}) for u, var in uc.items(): if solver.Value(var) == 1: detail = f"upgrade: {u}" elif chosen == Action.RENOVATE: rc = b._renovate_choice.get((ci, t), {}) for ct, var in rc.items(): if solver.Value(var) == 1: detail = f"renovate -> {ct}" elif chosen == Action.LAUNCH: detail = "launch airship" action_label = chosen.value if chosen is not None else Action.IDLE.value if foreman_renov: rc = b._renovate_choice.get((ci, t), {}) tgt = next( (ct for ct, var in rc.items() if solver.Value(var) == 1), None) rdetail = f"renovate -> {tgt}" if tgt else "renovate" detail = f"{detail} + {rdetail}" if detail else rdetail if chosen is None or chosen == Action.IDLE: action_label = Action.RENOVATE.value # Net per-resource change attributed to this City this Turn: the # change produced by its Industry Action, plus any resources granted # to it by a governing Agent (Provisioner/Courier). deltas = {} for r in RESOURCES: total = 0 for expr in b.city_delta.get((ci, r, t), []): total += expr if isinstance(expr, int) else solver.Value(expr) if total: # Electrum deltas are accumulated in tenths. deltas[r] = (float(total) / ELECTRUM_SCALE if r == "electrum" else float(total)) for r, amt in extra.get((ci, t), {}).items(): deltas[r] = deltas.get(r, 0.0) + amt if deltas[r] == 0: del deltas[r] plan.append(CityTurnPlan( turn=t, city=city.name, action=action_label, detail=detail, governor=governor, overwork=overwork, deltas=deltas, )) def _unscale(r, raw): # The Electrum pool is stored in tenths; everything else is 1:1. return float(raw) / ELECTRUM_SCALE if r == "electrum" else float(raw) final_resources = { r: _unscale(r, solver.Value(b.final_amt[r])) for r in RESOURCES } conversions = [] for (r, t), var in sorted(b.trade_conv.items(), key=lambda kv: (kv[0][1], kv[0][0])): amt = solver.Value(var) if amt: conversions.append({"turn": t, "resource": r, "amount": amt}) optional_conversions = [] for meta, n in b._opt_conv_vars: count = int(solver.Value(n)) if count: optional_conversions.append({ "turn": meta["turn"], "from": meta["from"], "to": meta["to"], "from_amount": meta["from_amount"], "to_amount": meta["to_amount"], "count": count, }) resources_by_turn = [] for t in range(T): row = {"turn": t} for r in RESOURCES: row[r] = _unscale(r, solver.Value(b._resource_at(r, t))) resources_by_turn.append(row) return Solution( status=status_name, objective_value=solver.ObjectiveValue() / OBJ_SCALE, final_resources=final_resources, final_renown_total=int(solver.Value(b.renown_total)), start_resources={ r: (round(problem.start.get(r, 0) * ELECTRUM_SCALE) / ELECTRUM_SCALE if r == "electrum" else float(int(round(problem.start.get(r, 0))))) for r in RESOURCES}, plan=sorted(plan, key=lambda p: (p.turn, p.city)), trade_conversions=conversions, forced_conversions=list(b.conversions), optional_conversions=optional_conversions, resources_by_turn=resources_by_turn, ) # --------------------------------------------------------------------------- # # JSON interface # --------------------------------------------------------------------------- # def problem_from_dict(d: dict) -> Problem: cities = [] for c in d.get("cities", []): c = dict(c) c["type"] = CityType(c["type"]) if "forced_action" in c: c["forced_action"] = {int(k): v for k, v in c["forced_action"].items()} cities.append(City(**c)) agents = [] for a in d.get("agents", []): a = dict(a) if "forced_city" in a: a["forced_city"] = {int(k): v for k, v in a["forced_city"].items()} agents.append(Agent(**a)) obj_d = dict(d.get("objective", {})) if "terms" in obj_d: obj_d["terms"] = [ t if isinstance(t, ScoreTerm) else ScoreTerm(**t) for t in obj_d["terms"] ] obj = Objective(**obj_d) kwargs = {k: v for k, v in d.items() if k not in ("cities", "agents", "objective")} return Problem(cities=cities, agents=agents, objective=obj, **kwargs) def solution_to_dict(s: Solution) -> dict: out = asdict(s) return out def main(argv: list[str]) -> int: import argparse ap = argparse.ArgumentParser(description="Days Without Strife planner optimizer") ap.add_argument("input", nargs="?", help="input JSON file (default stdin)") ap.add_argument("-o", "--output", help="output JSON file (default stdout)") ap.add_argument("--time", type=float, default=30.0, help="solver time limit (s)") args = ap.parse_args(argv) raw = open(args.input).read() if args.input else sys.stdin.read() problem = problem_from_dict(json.loads(raw)) sol = solve(problem, max_time_seconds=args.time) out = json.dumps(solution_to_dict(sol), indent=2) if args.output: with open(args.output, "w") as f: f.write(out) else: print(out) return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))