The turn of the galaxy
In short: The turn of the galaxy is a sequence of phases in a fixed order; fleets move simultaneously (first all of them leave, then they arrive), events are drawn by weight, peace and war have a rule of at least MIN_PEACE turns, and the news and the grudge listen to the event bus without the publisher knowing about them.
Problem
Everyone acts at once, and the order decides
In a turn-based strategy game the player types end (or wait 3 for three turns in a row) and the whole galaxy moves: the rivals decide what to do, fleets set off, battles happen, planets produce, sometimes pirates show up. What is one key press for the player is a turn in the code: a sequence of phases in a fixed order.
The order is not decoration. Take the simplest example: three fleets in a chain, A→B, B→C, C→D, and all 3 ships are in A. If you process the orders one after another and move the ships in the same array, the ships get from A to D in one turn: the first order puts them in B, the second sees them there and sends them on to C, the third to D. A player who sees this understands at once that the fleets are teleports.
This lesson builds the turn of a small galaxy of three empires (A, B, C) and four systems in a line, with the same four ideas as in the canon:
- the phases of a turn and simultaneous movement;
- random events: how often they come and which ones more often;
- peace and war: a rule that does not allow declaring war and making peace every turn;
- the event bus: how the news, the grudge and the alerts learn what happened, without anyone calling anyone.
At the end there is a whole turn of phases that prints what happens, and a scene that shows it.
The idea
Phases and simultaneous movement
A turn is a function that calls phases one after another. In the canon they are: the AI’s diplomacy, its orders, the movement of all fleets, battles, conquest, production and research, the event, the end of the turn. In our small galaxy there are five: diplomacy, movement, battles, production, event.
Movement is split in two: first all the fleets leave their systems, only then do they arrive. Here are the two ways on the chain from above. An order is [from, to, ships]; the 3 ships are in system 0:
# [from, to, ships]ORDERS = [(0, 1, 3), (1, 2, 3), (2, 3, 3)]
def move_one_by_one(start, orders): ships = start[:] for source, target, count in orders: take = min(count, ships[source]) ships[source] -= take ships[target] += take return ships
def move_together(start, orders): ships = start[:] takes = [] # First every fleet leaves its system... for source, target, count in orders: take = min(count, ships[source]) ships[source] -= take takes.append(take) # ...and only then do they arrive. for i, order in enumerate(orders): ships[order[1]] += takes[i] return shipsstart = [3, 0, 0, 0]print("one by one: " + show(move_one_by_one(start, ORDERS)))print("at the same time: " + show(move_together(start, ORDERS)))one by one: 0 0 0 3at the same time: 0 3 0 0The one-by-one version is quietly wrong: the program does not crash, the fleets just become fast. The simultaneous one takes, for each order, min(ships of the order, ships in the system at the start) and removes them (leaving), and then adds them (arriving). The second order (B→C) sees zero ships in B, because the first ones have not arrived yet, and moves nothing. That is why the chain is 0 3 0 0: the ships have only got as far as B.
start[:] is a copy of the list: without it the function would change the caller’s list.More precisely, if you are curious
The same rule applies to everything that happens “at once”: two empires swapping systems, two fleets passing each other. If you move them one by one, the second sees the world the first has already changed, and the result depends on the order of the players. The rule of two steps (take everything, then put everything back) is the same as swapping two variables with a helper: you need memory for the “old”.
In the canon the movement is even closer to ours, with one addition: a long journey continues by itself. The order remembers the final destination (dest), and after arrival a new order is recorded for the next step. So a fleet sent across five systems travels for five turns, one system per turn.
The idea
Random events by weight
On every turn there is a 40% chance that one event happens to a random empire. Which one? If all kinds had an equal chance, pirates would come on the first turn, when the empire does not have a single ship. So each kind has a weight (how common it is) and an earliest turn (from when it is allowed). Here is the canon’s table and the choice:
# id, weight, earliest turnEVENT_DEFS = [ ("meteor", 20, 1), ("vein", 15, 1), ("pirates", 20, 6), ("breakthrough", 15, 1), ("unrest", 10, 5), ("boom", 15, 1), ("flare", 10, 1), ("derelict", 8, 3)]
def pick_weighted(rng, weights): total = sum(weights) roll = rng.next_int(total) for i, weight in enumerate(weights): if roll < weight: return i roll -= weight return len(weights) - 1
def roll_event(rng, turn): if not rng.chance(EVENT_CHANCE): return None empire = rng.next_int(len(NAMES)) weights = [weight if turn >= min_turn else 0 for _, weight, min_turn in EVENT_DEFS] return (pick_weighted(rng, weights), empire)Choosing by weight is a “roulette”: a number from 0 up to the sum of the weights is drawn, and you walk along the kinds, subtracting the weight of each one, until the number is smaller than the current weight. A kind that is not allowed yet gets weight 0 and never comes up, and the sum shrinks, so the others become more likely. The program runs twelve turns and then a thousand more, to show how many times each kind comes up:
rng = Rng(7)for turn in range(1, 13): roll = roll_event(rng, turn) if roll is not None: print("turn " + str(turn) + ": " + EVENT_DEFS[roll[0]][0] + " → " + NAMES[roll[1]])
counts = [0] * len(EVENT_DEFS)for _ in range(1000): roll = roll_event(rng, 99) if roll is not None: counts[roll[0]] += 1print("Events in 1000 turns:")for i, (event_id, weight, _) in enumerate(EVENT_DEFS): print(event_id + " " + str(weight) + ": " + str(counts[i]))turn 1: boom → Lyra Unionturn 2: breakthrough → Vega Academyturn 3: vein → Vega Academyturn 4: boom → Vega Academyturn 8: pirates → Lyra Unionturn 9: breakthrough → Terran EmpireEvents in 1000 turns:meteor 20: 70vein 15: 53pirates 20: 78breakthrough 15: 58unrest 10: 30boom 15: 50flare 10: 28derelict 8: 28The pirates (earliest turn 6) only show up on turn 8. Over the thousand turns the pirates (78) and the meteors (70), with weight 20, come up most often, and the solar flare (28) and the derelict ships (28), with weights 10 and 8, least often. The counts are not equal to the weights, because it is random, but their order is the same. The chance of an event is 40% (EVENT_CHANCE), and the generator has a seed: the same input gives the same output in all four languages, so a recording of the game can be replayed.
More precisely, if you are curious
In the canon every event does something different: meteors take metal, a vein raises the richness of a planet, pirates fight the empire’s fleet (and resources are won or lost), a breakthrough gives science points, unrest lowers a mine level, a boom gives gas and crystal, a flare silences the labs for one turn, derelict ships give two frigates. Here they all are (the events region of the canon). Look at the end of rollEvent: the result is not written to the screen, it is published. What that means is in the card about the bus.
# The random events: every turn there is a 40% chance that one happens to a random empire. The kind is by# weights (only from its earliest turn on). Everything is published on the bus; the news, the statistics# and the grudges are subscribers, and the events know nothing about them.EVENT_CHANCE = 40
def pick_empire_planet(world, empire): items = owned_planets(world, empire.id) if len(items) == 0: return None return items[world.rng.next_int(len(items))]
def pirate_fleet(world): strength = world.turn // 6 return [2 + strength, strength // 2, strength // 3, 0, 0]
def apply_event(world, kind, empire): rng = world.rng values = {'empire': empire_name(empire)} if kind == 'meteor': planet = pick_empire_planet(world, empire) if planet is None: return None loss = min_of(empire.res[0], 8 + rng.next_int(8)) empire.res[0] -= loss values['planet'] = planet_name(world, planet) values['n'] = loss elif kind == 'vein': planet = pick_empire_planet(world, empire) if planet is None: return None r = rng.next_int(RES_COUNT) tries = 0 while planet.rich[r] >= 5 and tries < RES_COUNT: r = (r + 1) % RES_COUNT tries += 1 if planet.rich[r] >= 5: return None planet.rich[r] += 1 values['planet'] = planet_name(world, planet) values['res'] = t('res.' + str(r)) elif kind == 'pirates': planet = pick_empire_planet(world, empire) if planet is None: return None system_id = planet.system_id mine = empty_counts() fleet = fleet_at(world, empire.id, system_id) for k in range(SHIP_KIND_COUNT): mine[k] = 0 if fleet is None else fleet.ships[k] mine[FORT_KIND] = forts_in(world, empire.id, system_id) result = run_battle(side_stats(empire), mine, side_stats(None), pirate_fleet(world)) for k in range(SHIP_KIND_COUNT): if fleet is not None: fleet.ships[k] = result.a[k] values['system'] = world.systems[system_id].name values['fleet'] = counts_text(pirate_fleet(world)) if result.winner == 'a': empire.res[0] += 20 empire.res[1] += 10 values['outcome'] = t('event.pirates.win') else: m = empire.res[0] // 5 g = empire.res[1] // 5 empire.res[0] -= m empire.res[1] -= g values['outcome'] = t('event.pirates.lose', {'m': m, 'g': g}) if not world.quiet: world.last.battle = battle_lines(values['system'], empire_name(empire), t('pirates'), result) world.stats.battles += 1 world.stats.lost += total_of(mine) - total_of(result.a) + total_of(pirate_fleet(world)) - total_of(result.b) elif kind == 'breakthrough': n = 6 + rng.next_int(7) empire.progress += n values['n'] = n elif kind == 'unrest': planet = pick_empire_planet(world, empire) if planet is None or planet.mines == 0: return None planet.mines -= 1 values['planet'] = planet_name(world, planet) elif kind == 'boom': empire.res[1] += 20 empire.res[2] += 10 elif kind == 'flare': empire.flare = 1 elif kind == 'derelict': planet = pick_empire_planet(world, empire) if planet is None: return None add_ships(world, empire.id, planet.system_id, [0, 0, 2, 0]) values['system'] = world.systems[planet.system_id].name return values
def roll_event(world): rng = world.rng if not rng.chance(EVENT_CHANCE): return living = [] for e in world.empires: if e.alive: living.append(e) if len(living) == 0: return empire = living[rng.next_int(len(living))] allowed = [] weights = [] for definition in EVENT_DEFS: if world.turn >= definition.min_turn: allowed.append(definition) weights.append(definition.weight) definition = allowed[pick_weighted(rng, weights)] values = apply_event(world, definition.id, empire) if values is None: return world.stats.events += 1 world.bus.publish(SimpleNamespace(type='event', kind=definition.id, empire=empire.id, vars=values))
# The subscribers: the news for the player, the grudges after battles and conquest.def add_news(world, text): world.news.append(str(world.turn) + ': ' + text) while len(world.news) > 60: world.news.pop(0) world.turn_lines.append(text)
def install_subscribers(world, quiet): bus = world.bus
def involves_human(a, b): return world.empires[a].human or world.empires[b].human
def grudge_after_battle(e): if e.lost_a > 0: world.empires[e.b].grudge[e.a] += 1 if e.lost_b > 0: world.empires[e.a].grudge[e.b] += 1
def grudge_after_conquest(e): if e.from_ >= 0: world.empires[e.from_].grudge[e.empire] += 3
# A signal to the player: something happened at which "wait" stops. def alert_battle(e): if involves_human(e.a, e.b): world.alert += 1
def alert_conquest(e): if world.empires[e.empire].human or (e.from_ >= 0 and world.empires[e.from_].human): world.alert += 1
def alert_war(e): if involves_human(e.a, e.b): world.alert += 1
def alert_siege(e): if world.empires[e.from_].human: world.alert += 1
def alert_offer(e): if world.empires[e.to].human: world.alert += 1
bus.subscribe('battle', grudge_after_battle) bus.subscribe('conquest', grudge_after_conquest) bus.subscribe('battle', alert_battle) bus.subscribe('conquest', alert_conquest) bus.subscribe('war', alert_war) bus.subscribe('siege', alert_siege) bus.subscribe('offer', alert_offer) if quiet: return
def news_battle(e): if involves_human(e.a, e.b): for line in e.lines: add_news(world, line) else: add_news(world, t('news.battle', { 'system': world.systems[e.system_id].name, 'a': empire_name(world.empires[e.a]), 'b': empire_name(world.empires[e.b]), 'la': e.lost_a, 'lb': e.lost_b, }))
def news_colony(e): add_news(world, t('news.colony', {'empire': empire_name(world.empires[e.empire]), 'planet': planet_name(world, world.planets[e.planet_id])}))
def news_tech(e): if world.empires[e.empire].human: add_news(world, t('news.tech', {'tech': nm(TECHS[e.tech_id])}))
def news_war(e): add_news(world, t('news.war', {'a': empire_name(world.empires[e.a]), 'b': empire_name(world.empires[e.b])}))
def news_peace(e): add_news(world, t('news.peace', {'a': empire_name(world.empires[e.a]), 'b': empire_name(world.empires[e.b])}))
def news_offer(e): add_news(world, t('news.offer', {'empire': empire_name(world.empires[e.from_]), 'letter': world.empires[e.from_].def_.letter}))
def news_siege(e): add_news(world, t('news.siege', {'empire': empire_name(world.empires[e.empire]), 'planet': planet_name(world, world.planets[e.planet_id])}))
def news_conquest(e): add_news(world, t('news.conquest', {'empire': empire_name(world.empires[e.empire]), 'planet': planet_name(world, world.planets[e.planet_id])}))
def news_eliminated(e): add_news(world, t('news.eliminated', {'empire': empire_name(world.empires[e.empire])}))
def news_event(e): add_news(world, t('event.' + e.kind, e.vars))
bus.subscribe('battle', news_battle) bus.subscribe('colony', news_colony) bus.subscribe('tech', news_tech) bus.subscribe('war', news_war) bus.subscribe('peace', news_peace) bus.subscribe('offer', news_offer) bus.subscribe('siege', news_siege) bus.subscribe('conquest', news_conquest) bus.subscribe('eliminated', news_eliminated) bus.subscribe('event', news_event)The idea
Peace and war: a rule against chaos
Diplomacy in the game is a relation between two empires: a pair of empires is at war or at peace (war[a][b]). What is hard about it? Without a rule an empire would declare war on one turn, make peace on the next and declare war again right after. So:
- peace lasts at least MIN_PEACE =
4turns before a new war (turn − since < MIN_PEACE; at the start of the game the rule does not apply, so that someone can begin); - a peace offer is allowed at the earliest two turns after the change;
- the AI accepts peace according to its wish: how much weaker it is than the other, what its character is, how many other wars it is fighting and how resentful it is. From 90 up it accepts.
In the program the wish is simplified: character + (150 − ratio of ships) / 2 + other wars × 15 − grudge × 3, where the character is 40 for Orion (a warlord) and 120 for the other two (builders). A refusal raises the grudge, so the next attempt is harder. The script is the story of the scene: Orion (B) declares war on the Terran Empire (A) on turn 1, Lyra (C) on Orion on turn 3, and A offers peace:
class World: def __init__(self): self.turn = 1 self.ships = [[0] * 4 for _ in range(3)] self.war = [[0] * 3 for _ in range(3)] self.since = [[0] * 3 for _ in range(3)] self.grudge = [[0] * 3 for _ in range(3)]
def at_war(world, a, b): return world.war[a][b] == 1
def power(world, empire): return sum(world.ships[empire])
def set_stance(world, a, b, war): world.war[a][b] = war world.war[b][a] = war world.since[a][b] = world.turn world.since[b][a] = world.turn
def declare_war(world, a, b): if at_war(world, a, b): return "alreadyWar" # Peace lasts at least MIN_PEACE turns. if world.turn - world.since[a][b] < MIN_PEACE and world.turn > MIN_PEACE: return "tooSoon" set_stance(world, a, b, 1) return None
def make_peace(world, a, b): set_stance(world, a, b, 0) world.grudge[a][b] = 0 world.grudge[b][a] = 0
def peace_wish(world, me, other): # How much weaker the empire is than the other one. ratio = min(150, power(world, me) * 100 // (power(world, other) + 1)) other_wars = 0 for x in range(3): if x != me and x != other and at_war(world, me, x): other_wars += 1 return max(0, PEACE[me] + (150 - ratio) // 2 + other_wars * 15 - world.grudge[me][other] * 3)
def propose_peace(world, a, b): if not at_war(world, a, b): return "noWar" if world.turn - world.since[a][b] < 2: return "tooSoon" # From 90 up the empire accepts peace. if peace_wish(world, b, a) >= 90: make_peace(world, a, b) return None # A refusal makes the empire more resentful: the next peace is harder. world.grudge[b][a] += 1 return "refused"def step(world, turn, kind, a, b): world.turn = turn wish = peace_wish(world, b, a) if kind == "peace" else 0 error = declare_war(world, a, b) if kind == "war" else propose_peace(world, a, b) shown = " (wish " + str(wish) + ")" if kind == "peace" and error in (None, "refused") else "" print("turn " + str(turn) + ": " + kind + " " + NAMES[a] + " → " + NAMES[b] + ": " + (error or "ok") + shown)
world = World()world.ships[1][1] = 1world.grudge[1][0] = 1step(world, 1, "war", 1, 0)step(world, 2, "war", 1, 0)step(world, 2, "peace", 0, 1)step(world, 3, "war", 2, 1)step(world, 3, "peace", 0, 1)world.ships[1][1] = 0step(world, 4, "peace", 0, 1)step(world, 5, "war", 1, 0)step(world, 9, "war", 1, 0)turn 1: war Orion Hegemony → Terran Empire: okturn 2: war Orion Hegemony → Terran Empire: alreadyWarturn 2: peace Terran Empire → Orion Hegemony: tooSoonturn 3: war Lyra Union → Orion Hegemony: okturn 3: peace Terran Empire → Orion Hegemony: refused (wish 77)turn 4: peace Terran Empire → Orion Hegemony: ok (wish 124)turn 5: war Orion Hegemony → Terran Empire: tooSoonturn 9: war Orion Hegemony → Terran Empire: okTurn 2: Orion declares war on the same opponent again (alreadyWar), and A offers peace only one turn after the war (tooSoon). On turn 3 A offers peace, but Orion’s wish is 40 + 25 + 15 − 3 = 77 (character, one ship against none, one more war with Lyra, grudge 1) and it is below 90: refused, and the grudge becomes 2. On turn 4 Orion has no ships and its wish is 40 + 75 + 15 − 6 = 124: it accepts. Turn 5: Orion wants a new war, but the peace dates from turn 4, and 5 − 4 = 1 is less than 4: tooSoon. On turn 9 it is allowed (9 − 4 = 5).
def at_war(world, a, b):
return world.war[a][b] == 1
def power(world, empire):
return sum(world.ships[empire])
def set_stance(world, a, b, war):
world.war[a][b] = war
world.war[b][a] = war
world.since[a][b] = world.turn
world.since[b][a] = world.turn
def declare_war(world, a, b):
if at_war(world, a, b):
return "alreadyWar"
# Peace lasts at least MIN_PEACE turns.
if world.turn - world.since[a][b] < MIN_PEACE and world.turn > MIN_PEACE:
return "tooSoon"
set_stance(world, a, b, 1)
return None
def make_peace(world, a, b):
set_stance(world, a, b, 0)
world.grudge[a][b] = 0
world.grudge[b][a] = 0
def peace_wish(world, me, other):
# How much weaker the empire is than the other one.
ratio = min(150, power(world, me) * 100 // (power(world, other) + 1))
other_wars = 0
for x in range(3):
if x != me and x != other and at_war(world, me, x):
other_wars += 1
return max(0, PEACE[me] + (150 - ratio) // 2 + other_wars * 15 - world.grudge[me][other] * 3)
def propose_peace(world, a, b):
if not at_war(world, a, b):
return "noWar"
if world.turn - world.since[a][b] < 2:
return "tooSoon"
# From 90 up the empire accepts peace.
if peace_wish(world, b, a) 90:
make_peace(world, a, b)
return None
# A refusal makes the empire more resentful: the next peace is harder.
world.grudge[b][a] += 1
return "refused"The threshold is “from 90 up”, i.e. 90 is enough: >=. With > a wish of exactly 90 would be a refusal, while the rule in the game says the opposite.
More precisely, if you are curious
The result of these functions is a string or nothing: null if the change happened, and an error key (alreadyWar, tooSoon, refused, noWar) if it did not. The errors are keys, not finished sentences, so the screen can translate them (the game comes in four languages). That way the same function serves the player (the command peace B) and the AI.
The canon also has offers to the player: the AI cannot make peace with a human by itself, so it records an offer (offers) that is valid for three turns and publishes an “offer” event. Here is the whole diplomacy region:
# Diplomacy is peace or war between two empires. Peace holds for at least MIN_PEACE turns before# a new war. The artificial intelligence accepts peace by a measure: how much weaker it is, what its character is# and how bitter it is. An offer to the player is recorded and holds for three turns.def at_war(world, a, b): return world.empires[a].war[b] == 1
def set_stance(world, a, b, war): world.empires[a].war[b] = war world.empires[b].war[a] = war world.empires[a].since[b] = world.turn world.empires[b].since[a] = world.turn
def declare_war(world, a, b): if a == b: return 'dip.self' if not world.empires[b].alive: return 'dip.dead' if at_war(world, a, b): return 'dip.alreadyWar' if world.turn - world.empires[a].since[b] < MIN_PEACE and world.turn > MIN_PEACE: return 'dip.tooSoon' set_stance(world, a, b, 1) world.bus.publish(SimpleNamespace(type='war', a=a, b=b)) return None
def make_peace(world, a, b): set_stance(world, a, b, 0) world.empires[a].grudge[b] = 0 world.empires[b].grudge[a] = 0 world.bus.publish(SimpleNamespace(type='peace', a=a, b=b))
# The smallest distance in lanes between systems with planets of the two empires.def empire_distance(world, a, b): best = 99 for pa in world.planets: if pa.owner != a: continue dist = bfs(world.systems, pa.system_id).dist for pb in world.planets: if pb.owner == b and dist[pb.system_id] >= 0 and dist[pb.system_id] < best: best = dist[pb.system_id] return best
def power_ratio(world, a, b, top): return clamp(percent(empire_power(world, a) * 100, empire_power(world, b) * 100 + 100), 0, top)
# How much the empire wants peace with the other (0 and up; from 90 up it accepts).def peace_wish(world, me, other): e = world.empires[me] weaker = 150 - power_ratio(world, me, other, 150) other_wars = 0 for x in world.empires: if x.alive and x.id != me and x.id != other and e.war[x.id] == 1: other_wars += 1 wish = e.personality.peace + weaker // 2 + other_wars * 15 - e.grudge[other] * 3 return max_of(0, wish)
# How much the empire wants war with the other.def war_wish(world, me, other): e = world.empires[me] ratio = power_ratio(world, me, other, 300) wish = e.personality.attack // 2 if ratio > 100: wish += (ratio - 100) // 4 else: wish -= (100 - ratio) // 2 if empire_distance(world, me, other) <= 3: wish += 25 for x in world.empires: if x.alive and x.id != me and e.war[x.id] == 1: wish -= 30 return wish
def has_offer(world, src, to): for offer in world.offers: if offer.from_ == src and offer.to == to and world.turn - offer.turn <= 3: return True return False
# A peace offer from a to b. Returns an error key or None (peace is made).def propose_peace(world, a, b): if a == b: return 'dip.self' if not world.empires[b].alive: return 'dip.dead' if not at_war(world, a, b): return 'dip.noWar' if world.turn - world.empires[a].since[b] < 2: return 'dip.tooSoon' if world.empires[b].human: if has_offer(world, b, a): make_peace(world, a, b) return None if not has_offer(world, a, b): world.offers.append(SimpleNamespace(from_=a, to=b, turn=world.turn)) return 'dip.sent' if has_offer(world, b, a) or peace_wish(world, b, a) >= 90: make_peace(world, a, b) return None world.empires[b].grudge[a] += 1 return 'dip.refused'
# The diplomacy of the artificial intelligence at the end of the turn: peace when weaker, war when stronger.def ai_diplomacy(world, empire): for other in world.empires: if not other.alive or other.id == empire.id: continue if at_war(world, empire.id, other.id): if world.turn - empire.since[other.id] >= 3 and peace_wish(world, empire.id, other.id) >= 90: if other.human: if not has_offer(world, empire.id, other.id): world.offers.append(SimpleNamespace(from_=empire.id, to=other.id, turn=world.turn)) world.bus.publish(SimpleNamespace(type='offer', from_=empire.id, to=other.id)) else: propose_peace(world, empire.id, other.id) elif world.turn - empire.since[other.id] >= MIN_PEACE + 2 and world.turn > MIN_PEACE + 2: if war_wish(world, empire.id, other.id) >= 95 and world.rng.chance(40): declare_war(world, empire.id, other.id)The idea
An event bus: whoever publishes does not know who is listening
When war is declared, a lot has to happen: a news item has to be written, the grudge of the two empires has to grow, an alert has to light up for the player, the war has to be counted in the statistics. The simplest way is for declareWar to call all of that. Then declareWar knows about the news, the grudge, the alert and the statistics, and every new feature is one more line in it and one more reason for it to break. The bus reverses the direction: declareWar says “war has been declared between A and B” and is done. Whoever is interested has subscribed to “war”.
@dataclass(frozen=True)class Event: type: str a: int b: int place: int = 0 n: int = 0
class EventBus: def __init__(self): self.handlers = {}
def subscribe(self, event_type, handler): if event_type not in self.handlers: self.handlers[event_type] = [] self.handlers[event_type].append(handler)
def publish(self, event): if event.type not in self.handlers: return for handler in list(self.handlers[event.type]): handler(event)subscribe records a handler for a kind of event, publish calls all the handlers of that kind. A kind without subscribers is not an error: the event goes nowhere. Here is the publishing in the two places where it happens (declaring war and making peace). The only new thing is the line with publish:
def declare_war(world, a, b): if at_war(world, a, b): return "alreadyWar" # Peace lasts at least MIN_PEACE turns. if world.turn - world.since[a][b] < MIN_PEACE and world.turn > MIN_PEACE: return "tooSoon" set_stance(world, a, b, 1) world.bus.publish(Event("war", a, b)) return None
def make_peace(world, a, b): set_stance(world, a, b, 0) world.grudge[a][b] = 0 world.grudge[b][a] = 0 world.bus.publish(Event("peace", a, b))And here is the program. The news listen to all kinds, the grudge only to battles, and report also publishes the failed attempts (refused, tooSoon) as events:
world = World()world.ships[1][1] = 1world.grudge[1][0] = 1
def grudge(event): world.grudge[event.a][event.b] += 1 world.grudge[event.b][event.a] += 1
# The news listen to every kind of event, the grudge only to battles.for event_type in NEWS_TYPES: world.bus.subscribe(event_type, lambda event: print(news_text(event)))world.bus.subscribe("battle", grudge)
world.turn = 1report(world, declare_war(world, 1, 0), 1, 0)world.bus.publish(Event("battle", 1, 0, 1, 2))world.turn = 3report(world, propose_peace(world, 0, 1), 0, 1)report(world, declare_war(world, 1, 0), 1, 0)print("grudge " + NAMES[0] + " → " + NAMES[1] + ": " + str(world.grudge[0][1]))print("grudge " + NAMES[1] + " → " + NAMES[0] + ": " + str(world.grudge[1][0]))Orion Hegemony declares war on Terran Empire.Battle at Vesta: Orion Hegemony and Terran Empire lose ships (2 each).Orion Hegemony refuses peace with Terran Empire.Orion Hegemony is already at war with Terran Empire.grudge Terran Empire → Orion Hegemony: 1grudge Orion Hegemony → Terran Empire: 3The attempt on turn 1 succeeds, the battle is published by hand, the refusal on turn 3 is an event (refused), and the repeated war is alreadyWar. The grudge of A toward B is 1 (from the battle), and that of B toward A is 3: one old, one from the battle and one from the refusal (a refusal raises the grudge of the one who refuses toward the one who offered).
lambda); the EventBus class knows nothing about it except that it can call it.More precisely, if you are curious
Without a bus, everything new that has to learn about a war needs a change in declareWar. With the bus, a new subscriber is just a new line when the world is created, and the war function is not touched. This is the observer pattern from the design patterns module.
A small but important detail: publish loops over a copy of the list of subscribers. A handler may subscribe to another event while it is being called; if the list changes during the loop, subscribers are skipped or repeated. The copy makes the rule clear: a new subscriber hears from the next publication on.
The order of the subscribers is the order of subscribe. Here it does not matter (the news and the grudge do not get in each other’s way), but if two subscribers depend on each other, the order will decide the outcome, and that is a reason to keep them independent.
The idea
The whole turn
Now everything together. The world is World: turn, phase, generator, bus, the fleets of the three empires (by system), metal, and the tables for war (war), since which turn (since) and grudge (grudge):
class World: def __init__(self, seed): self.turn = 1 self.phase = "" self.rng = Rng(seed) self.bus = EventBus() self.ships = [[3, 0, 0, 0], [0, 0, 0, 4], [0, 2, 0, 0]] self.metal = [50, 50, 50] self.war = [[0] * 3 for _ in range(3)] self.since = [[0] * 3 for _ in range(3)] self.grudge = [[0] * 3 for _ in range(3)]The orders and the diplomacy of each turn are a script (in the game the AI or the player decides them): an order is [turn, empire, from, to, ships], and a diplomacy entry is [turn, kind, from, to].
# [turn, empire, from, to, ships]ORDERS = [(1, 1, 3, 2, 4), (1, 0, 0, 1, 3), (2, 1, 2, 1, 4)]# [turn, kind (0 war, 1 peace), from, to]DIPLO = [(1, 0, 1, 0), (3, 0, 2, 1), (3, 1, 0, 1), (4, 1, 0, 1), (5, 0, 1, 0)]The phases are functions. Diplomacy carries out the script for the turn and publishes the errors. Movement is exactly the two steps from the first card. Battles walk over the systems: where two empires at war have ships, each loses as many as the smaller fleet, and a “battle” is published (the grudge subscribes to it):
def scripted_diplomacy(world): world.phase = "diplomacy" for turn, kind, a, b in DIPLO: if turn != world.turn: continue error = declare_war(world, a, b) if kind == 0 else propose_peace(world, a, b) report(world, error, a, b)def resolve_movement(world): world.phase = "movement" moving = [] # First every fleet leaves its system... for turn, empire, source, target, count in ORDERS: if turn != world.turn: continue take = min(count, world.ships[empire][source]) world.ships[empire][source] -= take if take > 0: moving.append((empire, source, target, take)) # ...and only then do they arrive. for empire, source, target, take in moving: world.ships[empire][target] += take world.bus.publish(Event("move", empire, source, target, take))def resolve_battles(world): world.phase = "battles" for system in range(4): for a in range(3): for b in range(a + 1, 3): if not at_war(world, a, b) or world.ships[a][system] == 0 or world.ships[b][system] == 0: continue lost = min(world.ships[a][system], world.ships[b][system]) world.ships[a][system] -= lost world.ships[b][system] -= lost world.bus.publish(Event("battle", a, b, system, lost))
def produce(world): world.phase = "production" for empire in range(3): world.metal[empire] += 10After them comes the event: a draw by the weights from the second card, with these effects: a meteor takes metal, a boom gives metal, derelict ships give two ships at the capital. At last runTurn calls them in order:
def roll_event(world): world.phase = "event" if not world.rng.chance(EVENT_CHANCE): return empire = world.rng.next_int(3) weights = [weight if world.turn >= min_turn else 0 for _, weight, min_turn in EVENT_DEFS] kind = EVENT_DEFS[pick_weighted(world.rng, weights)][0] if kind == "meteor": loss = min(world.metal[empire], 8 + world.rng.next_int(8)) world.metal[empire] -= loss world.bus.publish(Event("meteor", empire, 0, 0, loss)) elif kind == "boom": world.metal[empire] += 20 world.bus.publish(Event("boom", empire, 0, 0, 20)) else: world.ships[empire][HOME[empire]] += 2 world.bus.publish(Event("derelict", empire, 0, HOME[empire], 2))def run_turn(world): # The phases run in this order, and the order matters. scripted_diplomacy(world) resolve_movement(world) resolve_battles(world) produce(world) roll_event(world) world.turn += 1The program subscribes the news (they put the phase in which the event happened in front of every line) and the grudge, runs five turns and prints the result. The lines are exactly the news of the game:
world = World(7)
def grudge(event): world.grudge[event.a][event.b] += 1 world.grudge[event.b][event.a] += 1
def news(event): print(" [" + world.phase + "] " + news_text(event))
world.bus.subscribe("battle", grudge)for event_type in NEWS_TYPES: world.bus.subscribe(event_type, news)
for _ in range(5): print(fill("-- turn {0} --", world.turn)) run_turn(world)print("-- end --")for empire in range(3): print(fill("{0}: ships {1}, metal {2}", NAMES[empire], power(world, empire), world.metal[empire]))-- turn 1 -- [diplomacy] Orion Hegemony declares war on Terran Empire. [movement] Orion Hegemony sends a fleet from Brisa to Altan (ships: 4). [movement] Terran Empire sends a fleet from Kora to Vesta (ships: 3). [event] Meteor shower: Lyra Union loses 14 metal.-- turn 2 -- [movement] Orion Hegemony sends a fleet from Altan to Vesta (ships: 4). [battles] Battle at Vesta: Terran Empire and Orion Hegemony lose ships (3 each).-- turn 3 -- [diplomacy] Lyra Union declares war on Orion Hegemony. [diplomacy] Orion Hegemony refuses peace with Terran Empire. [battles] Battle at Vesta: Orion Hegemony and Lyra Union lose ships (1 each).-- turn 4 -- [diplomacy] Peace between Terran Empire and Orion Hegemony. [event] Meteor shower: Lyra Union loses 8 metal.-- turn 5 -- [diplomacy] Orion Hegemony cannot change its relations with Terran Empire yet: too soon. [event] Orion Hegemony finds derelict ships at Brisa (ships: 2).-- end --Terran Empire: ships 0, metal 100Orion Hegemony: ships 2, metal 100Lyra Union: ships 1, metal 78Read the output as a story. Turn 1: Orion declares war, the two fleets (Orion’s and the Terran Empire’s) set off simultaneously, and Lyra loses metal to a meteor. Turn 2: Orion’s fleet arrives at Vesta, where the Terran Empire already is, and both lose 3 ships. Turn 3: Lyra joins the war and in the battle at Vesta they lose 1 each; before that, A offers peace and is refused. Turn 4: Orion has no ships, its wish is 124 and peace is made. Turn 5: a new war is “too soon”, and Orion finds two derelict ships at its capital.
See it
One turn of the galaxy
Here is the same thing turn by turn. At the top the phase in which something happened lights up. In the middle are the systems and the fleets of the three empires: during movement all fleets take off at once and only then land. Below them you can see whether each pair is at war, and at the very bottom the bus: the published event lights up the “news” subscriber, and if it is a battle, “grudge” as well.
def run_turn(world): # The phases run in this order, and the order matters. scripted_diplomacy(world) resolve_movement(world) resolve_battles(world) produce(world) roll_event(world) world.turn += 1
class EventBus: def __init__(self): self.handlers = {}
def subscribe(self, event_type, handler): if event_type not in self.handlers: self.handlers[event_type] = [] self.handlers[event_type].append(handler)
def publish(self, event): if event.type not in self.handlers: return for handler in list(self.handlers[event.type]): handler(event)Step by step, in words
- A = Terran Empire, B = Orion Hegemony, C = Lyra Union. Four systems in a line; the turn goes through the phases from top to bottom.
- turn 1 · diplomacy: Orion Hegemony declares war on Terran Empire.
- turn 1 · movement: Orion Hegemony sends a fleet from Brisa to Altan (ships: 4). Terran Empire sends a fleet from Kora to Vesta (ships: 3).
- turn 1 · event: Meteor shower: Lyra Union loses 14 metal.
- turn 2 · movement: Orion Hegemony sends a fleet from Altan to Vesta (ships: 4).
- turn 2 · battles: Battle at Vesta: Terran Empire and Orion Hegemony lose ships (3 each).
- turn 3 · diplomacy: Lyra Union declares war on Orion Hegemony. Orion Hegemony refuses peace with Terran Empire.
- turn 3 · battles: Battle at Vesta: Orion Hegemony and Lyra Union lose ships (1 each).
- turn 4 · diplomacy: Peace between Terran Empire and Orion Hegemony.
- turn 4 · event: Meteor shower: Lyra Union loses 8 metal.
- turn 5 · diplomacy: Orion Hegemony cannot change its relations with Terran Empire yet: too soon.
- turn 5 · event: Orion Hegemony finds derelict ships at Brisa (ships: 2).
- After 5 turns: Terran Empire: ships 0, metal 100 · Orion Hegemony: ships 2, metal 100 · Lyra Union: ships 1, metal 78
The code next to the scene is runTurn and EventBus from the program. The lines that are working at the moment light up.
The idea
In the game
What we built is the same as in the canon, only smaller. Here is the real turn (the turn region). Above runTurn are the movement, which makes the same two steps (leaving, arriving), the battles and the conquest of planets; runTurn calls the phases in a fixed order, and at the end of the turn the grudge decays (every fifth turn it drops by 1) and old peace offers are thrown away.
# One turn of the galaxy has phases: diplomacy of the artificial intelligence, its orders, movement of# all fleets at once, battles, conquest, production and science, an event, end of the turn.# Movement is simultaneous: first all ships leave their systems, only then do they arrive.def resolve_movement(world): moving = [] for empire in world.empires: if not empire.alive: continue for order in empire.orders: fleet = fleet_at(world, empire.id, order.from_) if fleet is None: continue ships = [0, 0, 0, 0] for k in range(SHIP_KIND_COUNT): ships[k] = min_of(order.ships[k], fleet.ships[k]) fleet.ships[k] -= ships[k] if total_of(ships) > 0: moving.append(SimpleNamespace(owner=empire.id, to=order.to, dest=order.dest, ships=ships)) empire.orders = [] for move in moving: add_ships(world, move.owner, move.to, move.ships) # the long way continues on its own: a new order from the place where they arrived if move.to != move.dest: world.empires[move.owner].orders.append(SimpleNamespace(from_=move.to, to=next_step(world.systems, move.to, move.dest), dest=move.dest, ships=list(move.ships))) remove_empty_fleets(world)
def has_units(world, owner, system_id): fleet = fleet_at(world, owner, system_id) return (fleet is not None and total_of(fleet.ships) > 0) or forts_now(world, owner, system_id) > 0
def resolve_battles(world): for system in world.systems: for a in range(len(world.empires)): for b in range(a + 1, len(world.empires)): if not world.empires[a].alive or not world.empires[b].alive or not at_war(world, a, b): continue if has_units(world, a, system.id) and has_units(world, b, system.id): fight_at(world, system.id, a, b) remove_empty_fleets(world)
# A planet falls when an enemy is at war with its owner, has ships in the system, and the owner has# neither ships nor surviving fortresses there, for SIEGE_TURNS turns in a row: the first turns are a siege,# the last one is the capture. The defender has time to bring a fleet. The buildings are destroyed.def resolve_conquest(world): for system in world.systems: for pid in system.planets: planet = world.planets[pid] if planet.owner < 0: continue owner = planet.owner attacker = -1 if not has_units(world, owner, system.id): for e in world.empires: if not e.alive or e.id == owner or not at_war(world, e.id, owner): continue fleet = fleet_at(world, e.id, system.id) if fleet is not None and total_of(fleet.ships) > 0: attacker = e.id break if attacker < 0: planet.siege = 0 continue planet.siege += 1 if planet.siege < SIEGE_TURNS: world.bus.publish(SimpleNamespace(type='siege', empire=attacker, from_=owner, planet_id=planet.id)) continue planet.siege = 0 planet.owner = attacker planet.mines = max_of(0, planet.mines - 1) planet.lab = 0 planet.yard = 0 planet.fort = 0 world.bus.publish(SimpleNamespace(type='conquest', empire=attacker, from_=owner, planet_id=planet.id)) for empire in world.empires: if not empire.alive or len(owned_planets(world, empire.id)) > 0: continue empire.alive = False kept = [] for fleet in world.fleets: if fleet.owner != empire.id: kept.append(fleet) world.fleets = kept empire.orders = [] world.bus.publish(SimpleNamespace(type='eliminated', empire=empire.id))
def run_turn(world): world.turn_lines = [] world.fort_left = {} for empire in world.empires: if empire.alive and not empire.human: ai_diplomacy(world, empire) for empire in world.empires: if empire.alive and not empire.human: ai_play(world, empire) resolve_movement(world) resolve_battles(world) resolve_conquest(world) for empire in world.empires: if not empire.alive: continue produce(world, empire) complete_research(world, empire) finish_queue(world) roll_event(world) check_victory(world) world.turn += 1 if world.turn % 5 == 0: for empire in world.empires: for i in range(len(empire.grudge)): empire.grudge[i] = max_of(0, empire.grudge[i] - 1) fresh = [] for offer in world.offers: if world.turn - offer.turn <= 3: fresh.append(offer) world.offers = freshAnd here is the bus that the news, the grudge and the alert to the player listen to. The events are one class with more fields (in JavaScript, an object), and the bus is the same:
# The event bus: whoever publishes does not know who listens. The news, the grudges of the empires and# the statistics subscribe on their own, and the battle only says "a fleet fought at this system".class EventBus: def __init__(self): self.handlers = {}
def subscribe(self, event_type, handler): if event_type not in self.handlers: self.handlers[event_type] = [] self.handlers[event_type].append(handler)
def publish(self, event): if event.type not in self.handlers: return for handler in list(self.handlers[event.type]): handler(event)The differences from our program are many, and that is why they matter:
- There is no script. Orders and diplomacy come from
aiPlayandaiDiplomacy(the lesson about the rivals) and from the player’s commands. - The galaxy is a graph (the lesson on generating it), not a line. An order remembers its final destination and continues by itself.
- The battle (
fightAt, the lesson about fleets) has ship roles and six rounds, not “each loses the smaller number”. There are also fortresses and the conquest of a planet (two sieges). - The wish for peace uses the strength of the empire (not the number of ships), its character and its wish for war (
warWish). An offer to the player is recorded for three turns. - Production here is a flat 10 metal per empire; in the canon it depends on planets, mines and technologies.
- Events are eight kinds, with more varied effects (pirates fight with the same battle model).
- The bus carries other kinds too: conquest, siege, a technology discovered, a colony, a destroyed empire. The news and the alert to the player listen to them, and
waitstops when the alert grows.
Try it
Predict the output
Two fleets: the first order sends the 3 ships from system 0 to 1, the second sends 2 ships from system 1 to 2. The orders are processed simultaneously. What does the program print?
def move_together(start, orders): ships = start[:] takes = [] # First every fleet leaves its system... for source, target, count in orders: take = min(count, ships[source]) ships[source] -= take takes.append(take) # ...and only then do they arrive. for i, order in enumerate(orders): ships[order[1]] += takes[i] return ships
def show(ships): return " ".join(str(n) for n in ships)
orders = [(0, 1, 3), (1, 2, 2)]print(show(move_together([3, 0, 0, 0], orders)))First everyone leaves: the first order takes 3 ships from system 0, and the second sees zero ships in system 1 (the first ones have not arrived yet) and takes 0. Then they arrive: 3 ships in system 1, 0 in system 2. The second option is the result with orders one after another: 2 of the arrived ships leave again the same turn; the third is the initial state, without movement.
Try it
Find the bug
A colleague “fixed” the simultaneous movement, but with it the ships still pass through several systems per turn:
start = [3, 0, 0, 0]print("at the same time: " + show(move_together(start, ORDERS)))at the same time: 0 0 0 3The result is the same as with one-by-one movement. Which line sends the ships to the destination too early?
Pick the line with the bug.
The ships arrive (+= at the destination) already in the loop for leaving. When the second order comes, the ships of the first are already there and it takes them. Correct: in the first loop they are only taken (-=) and remembered in takes; arriving is a second loop, after everyone has left.
More precisely
More precisely: order, fairness and chance
The order of the players. With simultaneous movement the order of the orders does not change the result, because everyone first takes from the start of the turn. With battles, however, there is an order: system by system, pair by pair (a < b). If three empires at war are in one system, the first pair in the order fights first and the survivors go on to the next. The order is fixed, so the game stays repeatable.
The phases are a contract. Diplomacy comes before movement, because then a newly declared war can start a battle in the same turn. Production comes after conquest: a planet taken this turn does not produce for its old master. The event is almost last and works with the world as it has become (the meteor takes from metal that has already been produced). Swap two phases and the game stays “the same”, but the outcome is different. That is why the order is in one place (runTurn) and reads like a list.
Chance and repeatability. Everything random (the chance of an event, which empire, which kind, how much metal) goes through one generator with a seed. If two phases swap places and both draw numbers, the sequence is different. In our program only rollEvent uses the generator, so the output is stable; in the canon diplomacy and the AI also draw numbers, and the order of the phases decides who gets which number.
What is not for the bus. The bus is for news: “this happened”. It is not for commands (“do this”): the publisher does not know whether anyone heard and does not wait for an answer. Publish facts, not orders.
Exercise
An alert for the player
Solve the task in your editor. Start from the program for the whole turn.
Alert. In the game wait stops as soon as something happens to the player. Add an “alert” subscriber that counts the events war, peace and battle in which empire 0 (the player) takes part, as a or as b. After the summary the program prints one more line with the alerts (alerts: …). Do not touch declareWar, makePeace and resolveBattles: the subscriber is just one new line when the world is created and one function.
The output is (the last line is the new one):
-- turn 1 -- [diplomacy] Orion Hegemony declares war on Terran Empire. [movement] Orion Hegemony sends a fleet from Brisa to Altan (ships: 4). [movement] Terran Empire sends a fleet from Kora to Vesta (ships: 3). [event] Meteor shower: Lyra Union loses 14 metal.-- turn 2 -- [movement] Orion Hegemony sends a fleet from Altan to Vesta (ships: 4). [battles] Battle at Vesta: Terran Empire and Orion Hegemony lose ships (3 each).-- turn 3 -- [diplomacy] Lyra Union declares war on Orion Hegemony. [diplomacy] Orion Hegemony refuses peace with Terran Empire. [battles] Battle at Vesta: Orion Hegemony and Lyra Union lose ships (1 each).-- turn 4 -- [diplomacy] Peace between Terran Empire and Orion Hegemony. [event] Meteor shower: Lyra Union loses 8 metal.-- turn 5 -- [diplomacy] Orion Hegemony cannot change its relations with Terran Empire yet: too soon. [event] Orion Hegemony finds derelict ships at Brisa (ships: 2).-- end --Terran Empire: ships 0, metal 100Orion Hegemony: ships 2, metal 100Lyra Union: ships 1, metal 78alerts: 3More tasks
- Statistics. Add one more subscriber that counts the battles and the total of ships lost (each battle loses
non each of the two sides). It prints them after the alerts. - Truce. Lower
MIN_PEACEto 2 and run the program. Which lines of the output change, and why is turn 5 no longer “too soon”? - A new kind of event. Add
plaguewith earliest turn 4 and weight 10: the empire loses 5 metal. Which places do you change: the table,rollEvent,newsText, the list of kinds for the news? And the bus?
Test
Test yourself
Related lessons
Section titled "Related lessons"- Builds on: Lists: append, remove and sort · dict and set · Lambdas · dataclass, match and patterns
- Tasks on this topic: Epic Cosmos: the empire