ajurna/pysatcalc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

testing

Browse Source
This commit is contained in:
Ajurna
2025-11-11 19:27:55 +00:00
parent adee0f5ffb
commit 47bef8cbe4
3 changed files with 240 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

237
main.py
View File

@@ -1,7 +1,7 @@
from collections import defaultdict, deque
from enum import Enum
from math import ceil
from typing import Dict, List, Tuple, Optional, Any
from typing import Dict, List, Tuple, Optional, Any, Deque
import networkx as nx
from flask import Flask, render_template, request
from pydantic import BaseModel, Field
@@ -29,13 +29,45 @@ def _slugify(name: str) -> str:
prev_us = False
return ''.join(out).strip('_')
class Production(BaseModel):
recipe: Recipe
class ProductionLink(BaseModel):
production: Production
quantity: float
class Production(BaseModel):
recipe: Recipe
production_level: float
ingress: Dict[Items, ProductionLink] = Field(default_factory=dict)
egress: Dict[Items, ProductionLink] = Field(default_factory=dict)
@property
def inputs(self):
return {
item: quantity*self.production_level for item, quantity in self.recipe.inputs.items()
}
@property
def outputs(self):
return {
item: quantity*self.production_level for item, quantity in self.recipe.outputs.items()
}
def demand_satisfied(self):
if self.production_level <= 0:
return True
demand = self.recipe.inputs
for item in self.ingress:
demand[item] -= self.ingress[item].quantity
return all(demand[item] <= 0 for item in demand)
def remaining_output(self):
outputs = self.outputs
for item in self.egress:
outputs[item] -= self.egress[item].quantity
return outputs
class ProductionChain:
def __init__(self, preferred_recipes: Optional[Dict[Items, Recipe]] = None):
self.item_to_recipies: Dict[Items, list[Recipes]] = defaultdict(list)
def __init__(self):
self.recipe_map: Dict[Items, list[Recipes]] = defaultdict(list)
for r in Recipes:
for o in r.value.outputs:
self.item_to_recipies[o].append(r.value)
@@ -142,6 +174,194 @@ class ProductionChain:
# self.generate_graph(production_chain, production, raw_resources, excess, byproduct)
return production_chain
def build_chain(self, targets: Dict[Items, float]) -> Any:
if not targets:
return {}
demand = defaultdict(float)
item_to_production = defaultdict(list)
queue: Deque[Items] = deque()
for target in targets:
recipe = self.get_recipe(target)
p = Production(recipe=recipe, production_level=targets[target]/ recipe.outputs[target])
for inp in p.inputs:
queue.append(inp)
for output in p.outputs:
item_to_production[output].append(p)
while queue:
item = queue.popleft()
if item in item_to_production or item in RawResources:
continue
recipe = self.get_recipe(item)
p = Production(recipe=recipe, production_level=0)
for output in p.outputs:
item_to_production[output].append(p)
for inp in p.inputs:
queue.append(inp)
calc_queue = deque()
for item in item_to_production:
for production in item_to_production[item]:
if not production.demand_satisfied():
calc_queue.append(production)
print(calc_queue)
# print("item_to_production:", item_to_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
def generate_graph(self, production_chain, production, raw_resources, excess, byproduct):
nx_graph = nx.DiGraph()
for recipe in production_chain:
@@ -479,6 +699,7 @@ def create_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.SteelBeam: 3.0})
# app.run(host="0.0.0.0", port=5000, debug=True)
prod_chain = ProductionChain()
# prod_chain.compute_chain({Items.FusedModularFrame: 1.5})
prod_chain.build_chain({Items.IronIngot: 1.5})