105 lines
3.9 KiB
Python
105 lines
3.9 KiB
Python
"""
|
|
Production - The Idle Game: Automated DAG & Supply Chain Integrity Validator
|
|
Scans all sector manifests and verifies:
|
|
1. All input IDs reference existing, valid product IDs.
|
|
2. No circular / cyclic dependency loops exist (Directed Acyclic Graph invariant).
|
|
3. Economic margin scaling (Wholesale value > Raw input costs).
|
|
4. Tier ordering integrity (Downstream Tier >= Upstream Tier).
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
# Add backend directory to path
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
from catalog_loader import load_all_sectors_and_products
|
|
|
|
def validate_dag():
|
|
sectors, products = load_all_sectors_and_products()
|
|
errors = []
|
|
warnings = []
|
|
|
|
print(f"=== Validating Supply Chain DAG Integrity ===")
|
|
print(f"Discovered: {len(sectors)} Sectors, {len(products)} Commodities / Facilities.\n")
|
|
|
|
if not products:
|
|
errors.append("FATAL: No products discovered in catalog!")
|
|
return errors, warnings
|
|
|
|
# 1. Existence and Key Verification
|
|
for pid, p in products.items():
|
|
# Check required fields
|
|
for field in ['id', 'name', 'sector', 'tier', 'baseCycleTime', 'basePrice', 'outputAmount', 'baseFacilityCost']:
|
|
if field not in p:
|
|
errors.append(f"Product '{pid}' is missing required field '{field}'.")
|
|
|
|
# Check sector exists
|
|
sec_id = p.get('sector')
|
|
if sec_id not in sectors:
|
|
errors.append(f"Product '{pid}' references unknown sector '{sec_id}'.")
|
|
|
|
# Check inputs exist
|
|
inputs = p.get('inputs', [])
|
|
input_cost = 0.0
|
|
for inp in inputs:
|
|
inp_id = inp.get('id')
|
|
amt = inp.get('amount', 1)
|
|
if inp_id not in products:
|
|
errors.append(f"Product '{pid}' requires nonexistent input '{inp_id}'.")
|
|
else:
|
|
parent = products[inp_id]
|
|
input_cost += parent.get('basePrice', 0.0) * amt
|
|
# Tier sanity check
|
|
if parent.get('tier', 1) > p.get('tier', 1):
|
|
warnings.append(f"Tier Inversion: '{pid}' (Tier {p.get('tier')}) consumes higher-tier product '{inp_id}' (Tier {parent.get('tier')}).")
|
|
|
|
# Economic margin check (Wholesale price vs input cost)
|
|
output_val = p.get('basePrice', 0.0) * p.get('outputAmount', 1)
|
|
if inputs and output_val <= input_cost:
|
|
warnings.append(f"Negative Profit Margin: '{pid}' batch produces ${output_val:.2f} but consumes ${input_cost:.2f} in inputs.")
|
|
|
|
# 2. Circular Dependency (Cycle Detection via DFS)
|
|
visited = {} # 0 = unvisited, 1 = visiting (in recursion stack), 2 = visited
|
|
|
|
def has_cycle(curr_id, path):
|
|
visited[curr_id] = 1 # Mark as currently visiting
|
|
curr_p = products.get(curr_id, {})
|
|
for inp in curr_p.get('inputs', []):
|
|
nxt_id = inp.get('id')
|
|
if nxt_id not in products:
|
|
continue
|
|
if visited.get(nxt_id, 0) == 1:
|
|
# Cycle found!
|
|
cycle_str = " -> ".join(path + [nxt_id])
|
|
errors.append(f"CIRCULAR DEPENDENCY DETECTED: {cycle_str}")
|
|
return True
|
|
if visited.get(nxt_id, 0) == 0:
|
|
if has_cycle(nxt_id, path + [nxt_id]):
|
|
return True
|
|
visited[curr_id] = 2 # Done with this branch
|
|
return False
|
|
|
|
for pid in products:
|
|
if visited.get(pid, 0) == 0:
|
|
has_cycle(pid, [pid])
|
|
|
|
return errors, warnings
|
|
|
|
if __name__ == '__main__':
|
|
errs, warns = validate_dag()
|
|
if warns:
|
|
print("[WARNINGS]:")
|
|
for w in warns:
|
|
print(f" [WARN] {w}")
|
|
print()
|
|
|
|
if errs:
|
|
print("[ERRORS FOUND]:")
|
|
for e in errs:
|
|
print(f" [FAIL] {e}")
|
|
print(f"\nDAG Validation FAILED with {len(errs)} error(s).")
|
|
sys.exit(1)
|
|
else:
|
|
print("SUCCESS: DAG VALIDATION PASSED. Supply Chain is acyclic, consistent, and healthy!")
|
|
sys.exit(0)
|