Initial commit

2026-07-26 20:26:36 -04:00
commit 4a014aeb50
12 changed files with 1058 additions and 0 deletions
--- a/aoe/world.py
+++ b/aoe/world.py
@@ -0,0 +1,425 @@
+from collections import defaultdict, Counter
+
+from aoe.exceptions import *
+from aoe.costs import *
+from aoe.dependencies import *
+from aoe.gather import *
+from aoe.housing import *
+from aoe.production import *
+from aoe.times import *
+from aoe.types import *
+
+
+
+class World():
+
+
+    # Creating a unit is split into three stages:
+    #   queuing the unit
+    #   building the unit
+    #   adding the unit to the world
+    #
+    # Before queuing a thing, we:
+    #   check its dependencies (research, age, etc.)
+    #   pay its cost
+    # Once a thing is queued, it waits to be built.
+    #
+    # Before building a thing, we
+    #   check for housing capacity
+    #   check for available production buildings
+    # Once a thing is building, it waits to be added.
+    #
+    # We assume for now that anything that has built can be added.
+    #
+    # Note: queuing a unit is itself an event that gets added to the queue,
+    # and, if executed successfully, it adds another event to the queue that
+    # actually adds the unit to the world
+
+    # TODO farms, reseeding, etc.
+    #       _check_resource_building, _use_resource_building, _free_resource_building
+    # TODO sheep should die like farms
+    # TODO need some warning, penalty, etc. if villager is assigned to task
+    #      without appropriate building (e.g., mining with no camp)
+    # TODO need efficiency penalties on, e.g., lumber camp
+    #       too many vils/camp decreases rate
+    #       "stale" (old) camp decreases rate
+    # TODO research
+
+
+    MAX_WORLD_TIME = 3*60*60  # 3 hours
+
+
+    def _pay_cost(self, purchase):
+        for i in range(len(self.resources)):
+            if self.resources[i] < self.costs.costs[purchase][i]:
+                raise NotEnoughResourcesException
+        for i in range(len(self.resources)):
+            self.resources[i] -= self.costs.costs[purchase][i]
+
+
+    def _check_research_dependencies(self, research):
+        self._check_dependencies(self.research_deps.dependencies[research])
+
+
+    def _check_building_dependencies(self, building):
+        self._check_dependencies(self.building_deps.dependencies[building])
+
+
+    def _check_unit_dependencies(self, unit):
+        self._check_dependencies(self.unit_deps.dependencies[unit])
+
+
+    def _check_dependencies(self, dependencies):
+        for dep in dependencies:
+            if dep not in self.buildings and \
+                    dep not in self.research and \
+                    not self._is_sufficient_age(dep):
+                        raise MissingDependencyException
+
+
+    def _is_sufficient_age(self, age):
+        if type(age) == Ages and self.age >= age:
+            return True
+        return False
+
+
+    def _check_age_dependencies(self, age):
+        total_buildings = 0
+        for building in self.age_deps.dependencies[age].keys():
+            if self.has_building(building):
+                total_buildings += self.age_deps.dependencies[age][building]
+        if total_buildings < self.age_deps.dependencies_counts[age]:
+            raise MissingDependencyException
+
+
+    def buy_research(self, research):
+        self._check_research_dependencies(research)
+        self._pay_cost(research)
+        self._update_costs(research)
+
+
+    def _update_costs(self, research):
+        pass
+
+
+    def _update_gather_rate(self, task):
+        pass  # TODO
+        #num_farms = self.buildings.count(Buildings.FARM)
+        #num_workers = self.villager_tasks[task]
+        #villager_room = 
+
+
+    def assign_villager(self, previous_task, new_task, count=1):
+        if self.villager_tasks[previous_task] < count:
+            raise IllegalVillagerMoveException
+
+        self._check_gather(new_task, count)
+
+        self.villager_tasks[previous_task] -= count
+        self.villager_tasks[new_task] += count
+
+        self._update_gather_rate(new_task)
+
+
+    def _check_villager_move(self, villagers_from_count):
+        for task, count in villagers_from_count.items():
+            if self.villager_tasks[task] < count:
+                raise IllegalVillagerMoveException
+
+
+    def _move_villagers_to_build(self, villagers_from_count):
+        for task, count in villagers_from_count.items():
+            self.villager_tasks[task] -= count
+            self.villager_tasks[Tasks.BUILD] += count
+
+
+    def build(self, building, villagers_from, villagers_to):
+        if type(villagers_from) == Tasks:
+            villagers_from = [villagers_from]
+        if type(villagers_to) == Tasks:
+            villagers_to = [villagers_to]
+
+        if len(villagers_from) != len(villagers_to):
+            raise ConservationOfVillagersException
+
+        villagers_from_count = Counter(villagers_from)
+
+        self._check_villager_move(villagers_from_count)
+        self._check_building_dependencies(building)
+
+        self._pay_cost(building)
+        self._move_villagers_to_build(villagers_from_count)
+
+        villagers_to_count = Counter(villagers_to)
+
+        num_villagers_building = len(villagers_from)
+        solo_build_time = self.build_times.build_times[building]
+        build_time = 3*solo_build_time/(num_villagers_building + 2)
+        update_time = self.current_time + build_time
+
+        self.events[update_time].append((self._add_building,
+                                         (building, villagers_to_count,)))
+
+
+    def _add_building(self, building, villagers_to):
+
+        self.buildings.append(building)
+        self._free_production_building(building)
+
+        for task, count in villagers_to.items():
+            self.villager_tasks[Tasks.BUILD] -= count
+            self.villager_tasks[task] += count
+
+        if building == Buildings.HOUSE:
+            self._free_housing()
+        if building == Buildings.FARM:
+            # assumes a farm is worked nonstop after being built
+            update_time = self.current_time + times.exist_duration[Buildings.FARM]
+            self.events[update_time].append((self._remove_farm, ()))
+
+
+    def _remove_farm(self):
+        self.buildings.remove(Buildings.FARM)
+
+        num_farms = self.buildings.count(Buildings.FARM)
+        if self.villager_tasks[Tasks.FARM] > num_farms:
+            self.villager_tasks[Tasks.FARM] -= 1
+            self.villager_tasks[Tasks.IDLE] += 1
+
+
+    def advance_by_time(self, duration):
+        self._run_clock_until(self.time + duration)
+
+
+    def advance_until_time(self, end_time):
+        self._run_clock_until_time(end_time)
+
+
+    def _get_next_event_time(self):
+        try:
+            next_event_time = min(self.events.keys())
+        except ValueError:
+            next_event_time = World.MAX_WORLD_TIME
+
+        return next_event_time
+
+
+    def _run_clock_without_events(self, end_time):
+        self._gather_resources_for_duration(end_time - self.current_time)
+        self.current_time = end_time
+
+
+    def _run_clock_until_time(self, end_time):
+        while self.current_time < end_time:
+            next_event_time = self._get_next_event_time()
+            self._run_clock_without_events(min(next_event_time, end_time))
+            self._handle_events()
+
+
+    def _get_population_space(self):
+        return sum([self.housing.space[building] for building in self.buildings])
+
+
+    def _get_effective_population(self):
+        return self.population + self.units_being_built
+
+
+    def _free_housing(self):
+        pop_space = self._get_population_space()
+        original_queue_len = len(self.production_queue)
+
+        # Assumes each unit costs 1 pop
+        new_builds = []
+        i = 0
+        while (self._get_effective_population() < pop_space and
+               i < original_queue_len):
+
+            unit = self.production_queue[i]
+            if self._check_production(unit):
+                self._build_unit(unit)
+                new_builds.append(i)
+
+            i += 1
+
+        self.production_queue = [self.production_queue[i] 
+                                 for i in range(original_queue_len)
+                                 if i not in new_builds]
+
+
+    def _free_production_building_of_unit(self, unit):
+        building = self.production.building[unit]
+        self._free_production_building(building)
+
+
+    def _free_production_building(self, building):
+        self.production_available[building] += 1
+
+        # Start the next queued unit if one exists
+        ready_index = None
+        ready_unit = None
+        for i in range(len(self.production_queue)):
+            unit = self.production_queue[i]
+            if self.production.building[unit] == building:
+                ready_index = i
+                ready_unit = unit
+                break
+
+        if ready_index:
+            self.production_queue.remove(i)
+            self._build_unit(unit)
+
+
+    def _check_housing(self, unit):
+        # assumes for now that all units cost 1 pop
+        pop_space = sum([self.housing.space[building] for building in self.buildings])
+        if self._get_effective_population() >= pop_space:
+            raise YallGotHousedException
+
+
+    def _check_gather(self, task, count=1):
+        required = self.gather.required_gather_buildings[task]
+        have = set(self.buildings)
+
+        match = any(map(have.issuperset, required))
+
+        if not match:
+            raise GatherBuildingNotAvailableException
+
+        if task == task.FARM:
+            num_farms = self.buildings.count(Buildings.FARM)
+            if self.villager_tasks[task] + count > num_farms:
+                raise GatherBuildingNotAvailableException
+
+
+    def _check_production(self, unit):
+        building = self.production.building[unit]
+        if self.production_available[building] < 1:
+            raise ProductionBuildingNotAvailableException
+
+
+    def _use_housing_and_production(self, unit):
+        self._check_housing(unit)
+        self._check_production(unit)
+
+        building = self.production.building[unit]
+
+        self.production_available[building] -= 1
+        self.units_being_built += 1
+
+
+    def _kill_unit(self, unit, task=None):
+        if unit == Units.VILLAGER and not task:
+                raise IllegalVillagerMoveException
+
+        # assumes population of 1 for every unit
+        self.population -= 1
+
+        if unit == Units.VILLAGER:
+            self.villagers -= 1
+            self.villager_tasks[task] -= 1
+            self._update_gather_rate(task)
+
+        self._free_housing()
+
+
+    def _add_unit(self, unit, initial_task):
+        # assumes population of 1 for every unit
+        self.units_being_built -= 1
+        self.population += 1
+        self._free_production_building_of_unit(unit)
+
+        if unit == Units.VILLAGER:
+            self.villagers += 1
+            self.villager_tasks[Tasks.IDLE] += 1
+            self.assign_villager(Tasks.IDLE, initial_task)
+
+
+    def _build_unit(self, unit, initial_task):
+        try:
+            self._use_housing_and_production(unit)
+        except ProductionBuildingNotAvailableException:
+            self.production_queue.append(unit)
+        except YallGotHousedException:
+            pass
+        else:
+            update_time = self.current_time + self.build_times.build_times[unit]
+            self.events[update_time].append((self._add_unit, (unit, initial_task,)))
+
+
+    def queue_unit(self, unit, initial_task=None):
+        self._check_unit_dependencies(unit)
+        self._pay_cost(unit)
+
+        self._build_unit(unit, initial_task)
+
+
+    def queue_villager(self, initial_task):
+        self.queue_unit(Units.VILLAGER, initial_task)
+
+
+    def _gather_resources_for_duration(self, duration):
+        for task, count in self.villager_tasks.items():
+            gather_rates = self.gather.gather_rates[task]
+            for resource, rate in gather_rates.items():
+                self.resources[resource] += count*rate*duration
+
+
+    def _handle_events(self):
+        for func, args in self.events[self.current_time]:
+            func(*args)
+        del self.events[self.current_time]
+
+
+    def _update_age(self, new_age):
+        self.age = new_age
+
+
+    def increase_age(self):
+        if self.age == Ages.IMPERIAL:
+            raise Exception("no.")
+
+        new_age = self.age + 1
+
+        self._check_age_dependencies(new_age)
+        self._pay_cost(new_age)
+
+        update_time = self.current_time + self.build_times.build_times[new_age]
+        self.events[update_time].append((self._update_age, (new_age,)))
+
+
+    def _deal_with_civ(self):
+        pass  # deal with any civ-specific changes
+
+
+    def __init__(self, civ=None):
+        self.civ = civ
+
+        self.age = Ages.DARK
+        self.villagers = 3
+        self.population = 4
+        self.resources = [200, 200, 100, 200]
+        self.research = []
+        self.buildings = [Buildings.TOWNCENTER]
+
+        self.villager_tasks = {t: 0 for t in Tasks}
+        self.villager_tasks[Tasks.IDLE] = self.villagers
+
+        self.units_being_built = 0
+        self.production_available = {b: 0 for b in Buildings}
+        self.production_available[Buildings.TOWNCENTER] = 1
+        self.production_queue = []
+
+        self.current_time = 0
+        self.events = defaultdict(list)
+
+        self.build_times = Times()
+        self.costs = Costs()
+        self.building_deps = BuildingDependencies()
+        self.age_deps = AgeDependencies()
+        self.research_deps = ResearchDependencies()
+        self.unit_deps = UnitDependencies()
+        self.housing = Housing()
+        self.production = Production()
+        self.gather = Gather()
+
+        self._deal_with_civ()