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 from vanilla import Items, Machines, Recipes, Recipe, RawResources from rich import print 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.recipe_map: Dict[Items, list[Recipes]] = defaultdict(list) for r in Recipes: for o in r.value.outputs: self.recipe_map[o].append(r.value) def get_recipe(self, item: Items, recipe_map: Dict[Items, list[Recipes]]) -> Optional[Recipe]: # TODO: make logic for priority selection of recipes return recipe_map.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, self.recipe_map) 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) return production_chain def compute_chain(targets: Dict[Items, float], preferred_by_output: Optional[Dict[Items, str]] = None) -> Tuple[Dict[str, float], List[dict], Dict[str, float], Dict[str, float]]: """ Given desired output rates (item -> units/min), compute: - required raw input rates (raw item -> units/min) - a flat list of steps with building counts and per-building utilization - total production rates (echo of targets summed if multiple entries per item) - unused byproducts (items/min) produced by multi-output recipes but not consumed or targeted Uses the Recipes enum and Items objects. Now supports alternate recipes: when multiple recipes produce the same output item, a selection heuristic is used unless an explicit preference is configured. preferred_by_output: optional mapping from output Item -> recipe name to force selection for that item. """ # Build a mapping from output item -> list of recipes that produce it output_to_recipes: Dict[Items, List[Recipe]] = {} for r in Recipes: recipe = r.value for out_item in recipe.outputs.keys(): output_to_recipes.setdefault(out_item, []).append(recipe) # Optional explicit preferences: map output Item -> recipe name to prefer PREFERRED_RECIPE_BY_OUTPUT: Dict[Items, str] = preferred_by_output or {} # Heuristic to select a recipe when multiple alternatives exist def select_recipe_for(item: Items) -> Optional[Recipe]: candidates = output_to_recipes.get(item, []) if not candidates: return None # If explicit preference exists and matches a candidate, use it pref_name = PREFERRED_RECIPE_BY_OUTPUT.get(item) if pref_name: for c in candidates: if c.name == pref_name: return c # Otherwise pick the candidate with the highest per-building output for this item # Tie-breaker 1: smallest total input per unit of this output # Tie-breaker 2: deterministic by name def score(c: Recipe) -> Tuple[float, float, str]: per_build_out = c.outputs.get(item, 0.0) total_input = sum(c.inputs.values()) # Lower input per unit is better; we express as (total_input/per_build_out) # Protect against division by zero eff = float('inf') if per_build_out <= 0 else (total_input / per_build_out) return (per_build_out, -eff, c.name) return sorted(candidates, key=score, reverse=True)[0] # Aggregate demands for each item demand: Dict[Items, float] = {} def add_demand(item: Items, rate: float) -> None: if rate == 0: return demand[item] = demand.get(item, 0.0) + rate for item, rate in targets.items(): add_demand(item, rate) # Work lists steps: List[dict] = [] raw_requirements: Dict[str, float] = {} # Track produced and consumed rates to calculate unused byproducts produced: Dict[Items, float] = {} consumed: Dict[Items, float] = {} # Expand demanded craftable items into their inputs until only raw remain while True: craftable_item = next( (i for i, r in demand.items() if r > 1e-9 and i in output_to_recipes), None, ) if craftable_item is None: break needed_rate = demand[craftable_item] recipe = select_recipe_for(craftable_item) if recipe is None: # Should not happen because craftable_item is in output_to_recipes, # but guard anyway: treat as raw if selection failed. demand[craftable_item] = 0.0 raw_requirements[craftable_item.value.name] = raw_requirements.get(craftable_item.value.name, 0.0) + needed_rate continue per_building_output = recipe.outputs[craftable_item] # Buildings needed buildings = needed_rate / per_building_output if per_building_output > 0 else 0.0 buildings_ceiled = ceil(buildings - 1e-9) utilization = 0.0 if buildings_ceiled == 0 else buildings / buildings_ceiled # Record the step (as display-friendly strings) steps.append({ "item": craftable_item.value.name, "recipe": recipe.name, "building": recipe.building.value.name, "target_rate": needed_rate, "per_building_output": per_building_output, "buildings_float": buildings, "buildings": buildings_ceiled, "utilization": utilization, }) # Consume this demand and add input demands demand[craftable_item] -= needed_rate scale = buildings # exact fractional buildings to match demand exactly # Account for all outputs produced by this recipe at the chosen scale for out_item, out_rate_per_build in (recipe.outputs or {}).items(): produced[out_item] = produced.get(out_item, 0.0) + out_rate_per_build * scale # Add input demands and track consumption for in_item, in_rate_per_build in (recipe.inputs or {}).items(): rate_needed = in_rate_per_build * scale add_demand(in_item, rate_needed) consumed[in_item] = consumed.get(in_item, 0.0) + rate_needed # What's left in demand are raw items for item, rate in demand.items(): if rate <= 1e-9: continue raw_requirements[item.value.name] = raw_requirements.get(item.value.name, 0.0) + rate # Merge steps for same item/building merged: Dict[Tuple[str, str], dict] = {} for s in steps: key = (s["item"], s["building"]) if key not in merged: merged[key] = {**s} else: m = merged[key] m["target_rate"] += s["target_rate"] m["buildings_float"] += s["buildings_float"] m["buildings"] += s["buildings"] total_buildings = m["buildings"] m["utilization"] = 0.0 if total_buildings == 0 else m["buildings_float"] / total_buildings merged_steps = sorted(merged.values(), key=lambda x: (x["building"], x["item"])) # Echo total outputs (same as targets possibly aggregated), as item-name -> rate total_outputs: Dict[str, float] = {} for item, rate in targets.items(): total_outputs[item.value.name] = total_outputs.get(item.value.name, 0.0) + rate # Compute unused byproducts: produced but not consumed and not part of explicit targets unused_byproducts: Dict[str, float] = {} for item, qty_produced in produced.items(): qty_consumed = consumed.get(item, 0.0) qty_targeted = targets.get(item, 0.0) unused = qty_produced - qty_consumed - qty_targeted if unused > 1e-9: unused_byproducts[item.value.name] = unused return raw_requirements, merged_steps, total_outputs, unused_byproducts @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_ # 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.FusedModularFrame: 1.5})