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 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 def get_recipe(item: Items, recipe_map: Dict[Items, list[Recipes]]) -> Optional[Recipe]: return recipe_map.get(item, (None,))[0] def compute_chain2(targets: Dict[Items, float]) -> Any: if not targets: return {} recipe_map: Dict[Items, list[Recipes]] = defaultdict(list) for r in Recipes: for o in r.value.outputs: recipe_map[o].append(r.value) # print(recipe_map) demand = defaultdict(float) for target in targets: demand[target] = targets[target] production = defaultdict(float) # add demands to production # find a recipe for that demand # add inputs to demand queue = deque(targets) g = nx.DiGraph() production_queue = [] raw_resources = defaultdict(float) production_chain = defaultdict(float) while queue: item = queue.popleft() recipe = get_recipe(item, recipe_map) if recipe is None: raw_resources[item] += demand[item] - raw_resources[item] continue levels = [] for out, quantity in recipe.outputs.items(): if out in demand: target_quantity = demand[out] - production[out] if target_quantity > 0: levels.append(target_quantity / quantity) else: levels.append(0) production_level = max(levels) if max(levels) > 0 else 0 production_chain[recipe.name] = production_level for out, quantity in recipe.outputs.items(): production[out] += production_level * quantity for inp, quantity in recipe.inputs.items(): queue.append(inp) demand[inp] += production_level * quantity print(demand, production, raw_resources, production_chain) return production 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) compute_chain2({Items.ModularFrame: 45.0})