304 lines
11 KiB
Python
304 lines
11 KiB
Python
from collections import defaultdict, deque
|
|
from enum import Enum
|
|
from math import ceil
|
|
from typing import Dict, List, Tuple, Optional, Any
|
|
import networkx as nx
|
|
from flask import Flask, render_template, request
|
|
from pydantic import BaseModel, Field
|
|
from plus import Items, Machines, Recipes, Recipe, RawResources
|
|
# from vanilla import Items, Machines, Recipes, Recipe, RawResources
|
|
from rich import print
|
|
import matplotlib.pyplot as plt
|
|
app = Flask(__name__)
|
|
|
|
|
|
|
|
# Helpers to map item names to safe query parameter keys
|
|
def _slugify(name: str) -> str:
|
|
s = ''.join(ch.lower() if ch.isalnum() else '_' for ch in name)
|
|
# collapse consecutive underscores and trim
|
|
out = []
|
|
prev_us = False
|
|
for ch in s:
|
|
if ch == '_':
|
|
if not prev_us:
|
|
out.append('_')
|
|
prev_us = True
|
|
else:
|
|
out.append(ch)
|
|
prev_us = False
|
|
return ''.join(out).strip('_')
|
|
|
|
class Production(BaseModel):
|
|
recipe: Recipe
|
|
quantity: float
|
|
|
|
class ProductionChain:
|
|
def __init__(self):
|
|
self.item_to_recipies: Dict[Items, list[Recipes]] = defaultdict(list)
|
|
for r in Recipes:
|
|
for o in r.value.outputs:
|
|
self.item_to_recipies[o].append(r.value)
|
|
self.recipe_name_to_obj: Dict[str, Recipe] = {}
|
|
for r in Recipes:
|
|
self.recipe_name_to_obj[r.value.name] = r.value
|
|
|
|
self.demand: Dict[Items, float] = defaultdict(float)
|
|
self.production: Dict[Items, float] = defaultdict(float)
|
|
self.raw_resources: Dict[Items, float] = defaultdict(float)
|
|
self.production_chain: Dict[str, float] = defaultdict(float)
|
|
self.excess: Dict[Items, float] = defaultdict(float)
|
|
self.byproduct: Dict[Items, float] = defaultdict(float)
|
|
|
|
def get_recipe(self, item: Items) -> Optional[Recipe]:
|
|
# TODO: make logic for priority selection of recipes
|
|
return self.item_to_recipies.get(item, (None,))[0]
|
|
|
|
def calculate_excess(self, production: Dict[Items, float], demand: Dict[Items, float], raw_resources: Dict[Items, float]):
|
|
excess = defaultdict(float)
|
|
for item, quantity in demand.items():
|
|
total = production[item] - demand[item]
|
|
if total != 0:
|
|
if item in RawResources:
|
|
raw_resources[item] -= production[item]
|
|
else:
|
|
excess[item] = total
|
|
return excess
|
|
|
|
def calculate_byproduct(self, production: Dict[Items, float], demand: Dict[Items, float], raw_resources: Dict[Items, float]):
|
|
byproduct = defaultdict(float)
|
|
for item, quantity in production.items():
|
|
if item not in demand:
|
|
byproduct[item] = quantity
|
|
return byproduct
|
|
|
|
def compute_chain(self, targets: Dict[Items, float]) -> Any:
|
|
if not targets:
|
|
return {}
|
|
demand = defaultdict(float)
|
|
for target in targets:
|
|
demand[target] = targets[target]
|
|
|
|
queue = deque(targets)
|
|
production = defaultdict(float)
|
|
raw_resources = defaultdict(float)
|
|
production_chain = defaultdict(float)
|
|
while queue:
|
|
item = queue.popleft()
|
|
if item == Items.Silica:
|
|
print("silica")
|
|
recipe = self.get_recipe(item)
|
|
if item in RawResources:
|
|
raw_resources[item] += demand[item] - raw_resources[item]
|
|
continue
|
|
if item in production:
|
|
if production[item] == demand[item]:
|
|
continue
|
|
|
|
levels = []
|
|
for out, quantity in recipe.outputs.items():
|
|
if out in demand:
|
|
target_quantity = demand[out] - production[out]
|
|
levels.append(target_quantity / quantity)
|
|
|
|
production_level = max(levels)
|
|
production_chain[recipe.name] += production_level
|
|
for out, quantity in recipe.outputs.items():
|
|
production[out] += production_level * quantity
|
|
for byproduct, quantity in recipe.byproducts.items():
|
|
production[byproduct] += production_level * quantity
|
|
queue.append(byproduct)
|
|
for inp, quantity in recipe.inputs.items():
|
|
queue.append(inp)
|
|
demand[inp] += production_level * quantity
|
|
|
|
excess = self.calculate_excess(production, demand, raw_resources)
|
|
byproduct =self.calculate_byproduct(production, demand, raw_resources)
|
|
|
|
|
|
print("demand:", demand)
|
|
print("production:", production)
|
|
print("excess:", excess)
|
|
print("byproduct:", byproduct)
|
|
print("raw resources:", raw_resources)
|
|
print("production chain:", production_chain)
|
|
self.demand = demand
|
|
self.production = production
|
|
self.raw_resources = raw_resources
|
|
self.production_chain = production_chain
|
|
self.excess = excess
|
|
self.byproduct = byproduct
|
|
|
|
# self.generate_graph(production_chain, production, raw_resources, excess, byproduct)
|
|
return production_chain
|
|
|
|
def generate_graph(self, production_chain, production, raw_resources, excess, byproduct):
|
|
nx_graph = nx.DiGraph()
|
|
for recipe in production_chain:
|
|
nx_graph.add_node(recipe)
|
|
for inp in self.recipe_name_to_obj[recipe].inputs.keys():
|
|
if inp in RawResources:
|
|
nx_graph.add_edge(recipe, inp.name)
|
|
continue
|
|
input_recipe = self.get_recipe(inp)
|
|
nx_graph.add_edge(recipe, input_recipe.name)
|
|
nx.draw_spring(nx_graph, with_labels=True, font_weight='bold')
|
|
plt.show()
|
|
|
|
|
|
@app.route("/", methods=["GET"])
|
|
def index():
|
|
# Build selectable items list from Items enum (display names)
|
|
item_names = sorted([i.value.name for i in Items])
|
|
name_to_item = {i.value.name: i for i in Items}
|
|
|
|
result = None
|
|
error = None
|
|
selected_item = item_names[0] if item_names else ""
|
|
selected_rate = 60.0
|
|
|
|
# Read from query parameters for bookmarkable URLs
|
|
item_name = request.args.get("item") or selected_item
|
|
rate_str = request.args.get("rate")
|
|
selected_recipe = request.args.get("recipe") or ""
|
|
|
|
# Parse per-item recipe overrides from query params recipe_for_<slug(item)>
|
|
# Build slug -> Items map
|
|
slug_to_item: Dict[str, Items] = { _slugify(i.value.name): i for i in Items }
|
|
overrides: Dict[Items, str] = {}
|
|
for key, value in request.args.items():
|
|
if not key.startswith("recipe_for_"):
|
|
continue
|
|
if value is None or value == "":
|
|
continue
|
|
slug = key[len("recipe_for_"):]
|
|
item_enum = slug_to_item.get(slug)
|
|
if not item_enum:
|
|
continue
|
|
# Validate that the value is a valid recipe option for this item
|
|
candidates = []
|
|
for r in Recipes:
|
|
rec = r.value
|
|
if item_enum in rec.outputs:
|
|
candidates.append(rec.name)
|
|
if value in candidates:
|
|
overrides[item_enum] = value
|
|
|
|
rate = None
|
|
if rate_str is not None and rate_str != "":
|
|
try:
|
|
rate = float(rate_str)
|
|
if rate < 0:
|
|
raise ValueError
|
|
except (TypeError, ValueError):
|
|
error = "Please enter a valid non-negative number for rate (items per minute)."
|
|
rate = None
|
|
|
|
selected_item = item_name
|
|
if rate is not None:
|
|
selected_rate = rate
|
|
|
|
# Determine candidate recipes for the selected output item
|
|
recipe_options: List[str] = []
|
|
item_obj_for_options = name_to_item.get(selected_item)
|
|
if item_obj_for_options is not None:
|
|
for r in Recipes:
|
|
recipe = r.value
|
|
if item_obj_for_options in recipe.outputs:
|
|
recipe_options.append(recipe.name)
|
|
recipe_options.sort()
|
|
# Validate selected_recipe against available options
|
|
if selected_recipe not in recipe_options:
|
|
selected_recipe = ""
|
|
else:
|
|
selected_recipe = ""
|
|
|
|
# Build preferred map merging top-level selection and overrides
|
|
preferred: Optional[Dict[Items, str]] = None
|
|
if selected_recipe or overrides:
|
|
preferred = {}
|
|
preferred.update(overrides)
|
|
if selected_recipe and item_obj_for_options is not None:
|
|
preferred[item_obj_for_options] = selected_recipe
|
|
|
|
# Compute and also prepare per-item override options based on resulting chain
|
|
overrides_ui: List[dict] = []
|
|
|
|
if not error and item_name and rate is not None:
|
|
item_obj = name_to_item.get(item_name)
|
|
if item_obj is None:
|
|
error = "Unknown item selected."
|
|
else:
|
|
targets = {item_obj: rate}
|
|
raw, steps, outputs, unused = compute_chain(targets, preferred_by_output=preferred)
|
|
result = {
|
|
"targets": {item_name: rate},
|
|
"raw": raw,
|
|
"steps": steps,
|
|
"outputs": outputs,
|
|
"unused": unused,
|
|
}
|
|
|
|
# Collect unique output items from steps
|
|
unique_items = []
|
|
seen = set()
|
|
for s in steps:
|
|
item_nm = s.get("item")
|
|
if item_nm and item_nm not in seen:
|
|
seen.add(item_nm)
|
|
unique_items.append(item_nm)
|
|
# For each item, compute candidate recipes and current selection
|
|
for item_nm in unique_items:
|
|
item_enum2 = name_to_item.get(item_nm)
|
|
if not item_enum2:
|
|
continue
|
|
candidates = []
|
|
for r in Recipes:
|
|
rec = r.value
|
|
if item_enum2 in rec.outputs:
|
|
candidates.append({
|
|
"name": rec.name,
|
|
"building": rec.building.value.name,
|
|
})
|
|
if len(candidates) <= 1:
|
|
continue # only show when alternates exist
|
|
candidates.sort(key=lambda x: (x["name"]))
|
|
sel = None
|
|
if preferred and item_enum2 in preferred:
|
|
sel = preferred[item_enum2]
|
|
slug = _slugify(item_nm)
|
|
overrides_ui.append({
|
|
"item_name": item_nm,
|
|
"slug": slug,
|
|
"options": candidates,
|
|
"selected": sel or "",
|
|
})
|
|
|
|
# Build reset query (clear overrides)
|
|
reset_query = f"?item={selected_item}&rate={selected_rate}"
|
|
if selected_recipe:
|
|
reset_query += f"&recipe={selected_recipe}"
|
|
|
|
return render_template(
|
|
"index.html",
|
|
items=item_names,
|
|
result=result,
|
|
error=error,
|
|
selected_item=selected_item,
|
|
selected_rate=selected_rate,
|
|
recipe_options=recipe_options,
|
|
selected_recipe=selected_recipe,
|
|
overrides_ui=overrides_ui,
|
|
reset_query=reset_query,
|
|
)
|
|
|
|
|
|
def create_app():
|
|
return app
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# For local dev: python main.py
|
|
# app.run(host="0.0.0.0", port=5000, debug=True)
|
|
prod_chain = ProductionChain()
|
|
prod_chain.compute_chain({Items.ModularFrame: 1.5}) |