-
Notifications
You must be signed in to change notification settings - Fork 1
/
sim.py
297 lines (255 loc) · 13 KB
/
sim.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
"""Factorio simulation"""
# stdlib imports
import functools
import json
from collections import defaultdict, Counter
from types import SimpleNamespace
import math
# project imports
from errors import *
from files import load_files
from init import *
from utils import *
from craft import *
from data import *
class Sim():
def __init__(self, data_dict):
self.data = SimpleNamespace(**data_dict)
self.clear()
def launch(self):
if self.current_items['rocket-part'] >= 100:
self.current_items['rocket-part'] -= 100
return True
return False
# TODO: error message on crafting can return the full missing list
def craft(self, item, amount):
res, missing, available, msg = craftable(self, item, amount)
if res == 0:
missing[item] = amount
self.deduct_list(available)
self.place_in_inventory(self.data.recipes[item]['products'][0]['name'], amount)
time_spent = self.craft_time_list(missing)
print(time_spent)
# put excess production into player inventory based on bulk orders
self.grant_excess_production(missing)
if time_spent > 0:
self.next(time_spent)
return 0, None
elif res == 2:
return 1, f'crafting {amount} {item} failed, {msg}'
else:
return 1, f'something went wrong, {msg}'
def next(self, seconds, check_rates=False):
# simulate factory production, moving forwards in time
# depends on current state of inventory before next() was called.
# next(60) + next(60) != next(120), because state changes after each call
def produce(ci):
# potential = amount_of_machines * craft_speed * seconds // energy
def p_eq(amount_machines, amount_produced, speed, seconds, energy):
return amount_machines * speed * amount_produced * (seconds if energy == 1 else (seconds // energy))
def miner_potential(miner, item, amount, seconds):
return p_eq(amount, 1, self.data.mining_drills[miner]['mining_speed'], seconds, 1)
def assembler_potential(assembler, item, amount, seconds):
return p_eq(amount, self.data.recipes[item]['products'][0]['amount'], self.data.assemblers[assembler]['crafting_speed'], seconds, self.data.recipes[item]['energy'])
def furnace_potential(furnace, item, amount, seconds):
return p_eq(amount, 1, self.data.furnaces[furnace]['crafting_speed'], seconds, self.data.recipes[item]['energy'])
def machine_craft(item, num_produced, ci):
if item not in self.data.resources:
self.deduct_list(shopping_list(self.data.recipes, {item: num_produced}), ci)
self.place_in_inventory(self.data.recipes[item]['products'][0]['name'], num_produced, ci)
else:
self.place_in_inventory(item, num_produced, ci)
def miner_actual(item, potential):
return potential
def assembler_actual(item, potential):
# respect rate limits
if item in self.limited_items:
# should almost always be > 0 but if the limit was set after getting a lot of items we could go negative
potential = min(potential, max(0, self.limited_items[item] - ci[item]))
# find the bottleneck ingredient ratio and multiply by amount produced by recipe
# TODO: can we surface the bottleneck information to the player?
a = min([ci[x] // amount for x, amount in shopping_list(self.data.recipes, {item:1}).items()]) * self.data.recipes[item]['products'][0]['amount']
return min(a, potential)
def furnace_actual(item, potential):
return assembler_actual(item, potential)
prod_rates = defaultdict(lambda: defaultdict(int))
calc_actual = {0: miner_actual, 1: assembler_actual, 2: furnace_actual}
calc_potential = {0: miner_potential, 1: assembler_potential, 2: furnace_potential}
# core algo: for each machine: potential -> actual -> craft
# TODO: ordering of machines processed matters, because inventory is affected immediately
# players probably will probably want more control of this ordering
# having something deterministic will help with testing as well
# iterate through sorted keys - sort by priority!
for machine_item_key, amount in sorted(self.machines.items(), key=lambda item: int(item[0].split(':')[-1])):
item, machine, _ = machine_item_key.split(':')
key = self.data.machines[machine]
potential = calc_potential[key](machine, item, amount, seconds)
actual = calc_actual[key](item, potential)
prod_rates[item]['potential'] += potential
prod_rates[item]['actual'] += actual
machine_craft(item, actual, ci)
return prod_rates
# ---- end of produce() helper function
if check_rates:
ci = self.current_items.copy()
else:
# move time forwards and commit to item changes
self.game_time += seconds
ci = self.current_items
return produce(ci)
def place_machine(self, machine, item, amount=1):
res, msg = is_machine_compatible(self.data, machine, item)
if res != 0:
return res, f'failed to place {machine} producing {item}, {msg}'
if machine in self.data.assemblers:
if item not in self.current_recipes:
return 1, f'failed to place {machine} producing {item}, recipe locked'
# deduct the item *after* validation, so that we don't have to put it back if something goes wrong
res, msg = self.deduct_item(machine, amount)
if res != 0:
return res, f'failed to place {amount} of {machine}, {msg}'
priority = 0
self.machines[f'{item}:{machine}:{priority}'] += amount
return 0, None
def mine(self, resource, amount):
if resource in {'stone', 'coal', 'iron-ore', 'copper-ore', 'crude-oil'}:
self.place_in_inventory(resource, amount)
time_spent = self.data.resources[resource]['mineable_properties']['mining_time'] * amount
self.next(time_spent)
return 0, None
else:
return 1, f'{resource} cannot be mined'
def research(self, tech):
# research a given technology, raise exception if potions not available
# or given technology can not be researched yet
res, msg = self.researchable(tech)
if res == 0:
pl = get_potion_list(self.data.technology, tech)
self.deduct_list(pl)
self.current_tech.add(tech)
# unlock recipes
unlocked = []
for effect in self.data.technology[tech]['effects']:
if effect['type'] == 'unlock-recipe':
self.current_recipes.add(effect['recipe'])
unlocked.append(effect['recipe'])
unlocked_str = '\n'.join(unlocked)
msg = f'unlocked the following recipes:\n\n{unlocked_str}'
return res, msg
def check_item(self, item, amount, ci=None):
if ci == None:
ci = self.current_items
res = ci[item] >= amount
return res, max(0, amount - ci[item]), min(amount, ci[item])
def check_list(self, sh, ci):
def reduce_sh(accum, x):
res, missing, available = accum
r, m, av = self.check_item(x[0], x[1], ci)
res = r and res
if m > 0:
missing[x[0]] = m
if av > 0:
available[x[0]] = av
return res, missing, available
vals = [True, Counter(), Counter()]
res, missing, available = functools.reduce(lambda accum, x: reduce_sh(accum, x), sh.items(), vals)
return not res, missing, available
def deduct_item(self, item, amount, ci=None):
if ci == None:
ci = self.current_items
if ci[item] < amount:
return 1, f'player has < {amount} of {item} in inventory'
ci[item] -= amount
return 0, None
def deduct_list(self, sh, ci=None):
# NOTE: this function doesn't error check because it only gets called
# after some validation has already been done
if ci == None:
ci = self.current_items
for k, v in sh.items():
ci[k] -= v
def place_in_inventory(self, item, amount, ci=None):
if ci == None:
ci = self.current_items
# enfore integral system - avoid very real issues
ci[item] += int(amount)
def craft_time_list(self, craft_list):
time = 0
for name, amount in craft_list.items():
time += craft_time(self.data, name, amount)
print(time, name, amount)
return time
# TODO: is this correct?
def grant_excess_production(self, craft_list):
for name, amount in craft_list.items():
ratio = self.data.recipes[name]['main_product']['amount']
if amount % ratio != 0:
self.place_in_inventory(name, (ratio * (1 + (amount // ratio))) - amount)
def set_limit(self, item, amount):
self.limited_items[item] = amount
def set_machine_prio(self, machine, item, oldprio, newprio):
oldkey = f'{item}:{machine}:{oldprio}'
newkey = f'{item}:{machine}:{newprio}'
if self.machines[oldkey] > 0:
self.machines[oldkey] -= 1
self.machines[newkey] += 1
def preqs_researched(self, tech):
preq = self.data.technology[tech]['prerequisites']
return functools.reduce(lambda x, y: x and y in self.current_tech, preq, True)
def researchable(self, tech):
"""Decide whether a given technology can be researched or not"""
if tech not in self.data.technology:
return 1, f'researchable - {tech} could not be found in the list of tecnologies'
if tech in self.current_tech:
return 1, f'researchable - {tech} has already been researched'
pl = get_potion_list(self.data.technology, tech)
if not self.preqs_researched(tech):
return 1, f'researchable - one or more prerequisite technologies for {tech} have not been researched'
res, missing, _ = self.check_list(pl, self.current_items)
if res != 0:
return res, f'researchable - missing the potions required to research {tech}, {missing}'
return 0, None
def all_researchable(self):
return set(filter(lambda tech: tech not in self.current_tech and self.preqs_researched(tech), self.data.technology))
def clear(self):
self.game_time = 0
self.current_tech = get_starter_tech()
self.current_recipes = get_starter_recipes(self.data.recipes)
self.current_items = get_starter_inventory()
self.limited_items = dict()
self.machines = defaultdict(int)
def update_state(self, game_time, current_tech, current_recipes, current_items, machines, limited_items):
self.game_time = game_time
self.current_tech = current_tech
self.current_recipes = current_recipes
self.current_items = current_items
self.machines = machines
self.limited_items = limited_items
def serialize_state(self):
# every field is sorted so that this function is deterministic.
# the same actions performed in a simulation should produce the exact same
# save file every time! Important for running tests
def get_state():
return {
'game_time': self.game_time,
'current_tech': sorted(list(self.current_tech)),
'current_recipes': sorted(list(self.current_recipes)),
'current_items': dict(sorted(self.current_items.items())),
'machines': dict(sorted(self.machines.items())),
'limited_items': dict(sorted(self.limited_items.items()))
}
# Sort the outer dictionary and ensure inner dictionaries are sorted as well
s = {k: v if isinstance(v, (int, str, list, float)) else dict(sorted(v.items())) for k, v in get_state().items()}
return json.dumps(s, sort_keys=True)
def deserialize_state(self, s_json):
s = json.loads(s_json)
self.game_time = s['game_time']
self.current_tech = set(s['current_tech'])
self.current_recipes = set(s['current_recipes'])
self.current_items = defaultdict(int, s['current_items'])
self.machines = defaultdict(int, s['machines'])
self.limited_items = s['limited_items']
def production(self):
production = self.next(60, True)
data = [[k, v['actual'], v['potential'], self.current_items[k], (self.limited_items[k] if k in self.limited_items else '')] for k, v in production.items()]
return data