""" Production - The Idle Game: Core Simulation Backend Engine Authoritative offline tick progression, multi-tier DAG supply chains, and action validation. """ import math import time from catalog_loader import load_all_sectors_and_products # Dynamically loaded Product & Sector Registry from modular sector files SECTOR_CONFIGS, PRODUCT_CONFIGS = load_all_sectors_and_products() def refresh_catalog(): """Hot-reloads the product catalog if new sector files were added dynamically.""" global SECTOR_CONFIGS, PRODUCT_CONFIGS SECTOR_CONFIGS, PRODUCT_CONFIGS = load_all_sectors_and_products() UNIT_MILESTONES = [1, 10, 25, 50, 100, 200, 300, 400, 500, 750, 1000, 1500, 2000] BOARDROOM_PERK_COSTS = { 'supply_sync': [1, 3, 8, 20, 50], 'robotics_overclock': [2, 5, 12, 30, 75], 'logistics_efficiency': [2, 6, 15, 40, 100], 'government_subsidies': [3, 8, 20, 55, 120], 'angel_syndicate': [5, 15, 45, 120, 300], 'brand_legacy': [4, 10, 25, 70, 150] } def clean_round(val: float) -> float: if math.isnan(val) or math.isinf(val) or val < 0: return 0.0 return round(float(val), 2) def get_unit_milestone_multiplier(count: int) -> float: reached = sum(1 for m in UNIT_MILESTONES if count >= m) return float(math.pow(2.0, reached)) def get_initial_state() -> dict: state = { 'version': 1, 'companyName': 'Apex Production Industries', 'cash': 30.0, 'lifetimeRevenue': 0.0, 'valuation': 30.0, 'ventureShares': 0, 'totalVentureShares': 0, 'ipoCount': 0, 'reputation': 0, 'repAllocations': {'b2b': 0, 'clients': 0, 'brand': 0}, 'buyMode': '1', 'activeTab': 'production', 'selectedSector': 'all', 'selectedProductId': 'wheat', 'stats': { 'totalProduced': 0.0, 'contractsCompleted': 0, 'managersHired': 0 }, 'products': {}, 'boardroomPerks': {p: 0 for p in BOARDROOM_PERK_COSTS}, 'achievements': {}, 'activeContracts': [], 'lastTick': time.time() } for pid, p in PRODUCT_CONFIGS.items(): is_unlocked = p.get('unlockedDefault', False) state['products'][pid] = { 'unlocked': is_unlocked, 'stock': 0.0, 'lines': 1 if is_unlocked else 0, 'manager': False, 'logistics': 1 if is_unlocked else 0, 'logisticsEnabled': True, 'supplyDownstreamEnabled': True, 'productionTarget': 1.0, 'cycleProgress': 0.0, 'isBottlenecked': False, 'totalProduced': 0.0 } return state def calculate_shares_earned(lifetime_revenue: float) -> int: if lifetime_revenue < 100000.0: return 0 return int(math.floor(10.0 * math.sqrt(lifetime_revenue / 100000.0))) def process_offline_and_ticks(state: dict) -> dict: """Computes authoritative tick progress up to the current timestamp (capped at 8 hours).""" now = time.time() last = state.get('lastTick', now) elapsed = max(0.0, min(now - last, 8.0 * 3600.0)) state['lastTick'] = now if elapsed <= 0.1: return state perks = state.get('boardroomPerks', {}) rep_alloc = state.get('repAllocations', {}) if isinstance(state.get('repAllocations'), dict) else {} brand_rep = rep_alloc.get('brand', 0) clients_rep = rep_alloc.get('clients', 0) # Multipliers supply_yield_mult = (1.0 + 0.15 * perks.get('supply_sync', 0)) * (1.0 + 0.002 * clients_rep) robotics_speed_mult = (1.0 + 0.20 * perks.get('robotics_overclock', 0)) * (1.0 + 0.0015 * clients_rep) # Brand Premium: +0.25% wholesale price per point logistics_price_mult = (1.0 + 0.25 * perks.get('logistics_efficiency', 0)) * (1.0 + 0.0025 * brand_rep) # Simulate in reasonable sub-steps for chained DAG dependency sub_steps = max(1, min(int(elapsed), 120)) dt = elapsed / sub_steps for _ in range(sub_steps): for pid, p_cfg in PRODUCT_CONFIGS.items(): p_state = state.get('products', {}).get(pid) if not p_state or not p_state.get('unlocked'): continue lines = p_state.get('lines', 0) if lines <= 0: continue milestone_mult = get_unit_milestone_multiplier(lines) target_throttle = float(p_state.get('productionTarget', 1.0)) if target_throttle <= 0.0: p_state['isBottlenecked'] = False continue cycle_rate = (1.0 / p_cfg['baseCycleTime']) * milestone_mult * robotics_speed_mult * target_throttle inputs = p_cfg.get('inputs', []) if not inputs: # Raw extraction p_state['isBottlenecked'] = False p_state['cycleProgress'] = p_state.get('cycleProgress', 0.0) + (cycle_rate * dt) if p_state['cycleProgress'] >= 1.0: cycles = int(p_state['cycleProgress']) p_state['cycleProgress'] -= cycles produced = cycles * p_cfg['outputAmount'] * lines * supply_yield_mult p_state['stock'] = p_state.get('stock', 0.0) + produced p_state['totalProduced'] = p_state.get('totalProduced', 0.0) + produced else: # Manufacturing logic with clean batch cycle state needed = [(inp['id'], inp['amount'] * lines) for inp in inputs] def can_access_inputs(): for item_id, req in needed: upstream_state = state.get('products', {}).get(item_id, {}) if upstream_state.get('supplyDownstreamEnabled') is False: return False if upstream_state.get('stock', 0.0) < req: return False return True if p_state.get('cycleProgress', 0.0) <= 0.0: if can_access_inputs(): p_state['isBottlenecked'] = False for item_id, req in needed: state['products'][item_id]['stock'] = max(0.0, state['products'][item_id].get('stock', 0.0) - req) p_state['cycleProgress'] = min(1.0, cycle_rate * dt) else: p_state['isBottlenecked'] = True else: p_state['isBottlenecked'] = False p_state['cycleProgress'] = p_state.get('cycleProgress', 0.0) + (cycle_rate * dt) if p_state.get('cycleProgress', 0.0) >= 1.0: p_state['cycleProgress'] = 0.0 produced = p_cfg['outputAmount'] * lines p_state['stock'] = p_state.get('stock', 0.0) + produced p_state['totalProduced'] = p_state.get('totalProduced', 0.0) + produced # Check next batch if can_access_inputs(): for item_id, req in needed: state['products'][item_id]['stock'] = max(0.0, state['products'][item_id].get('stock', 0.0) - req) p_state['isBottlenecked'] = False else: p_state['isBottlenecked'] = True # Passive Logistics Sales with stock reservation for downstream factories logistics = p_state.get('logistics', 0) is_logistics_active = p_state.get('logisticsEnabled', True) is not False if is_logistics_active and logistics > 0 and p_state.get('stock', 0.0) > 0: reserved_stock = 0.0 if p_state.get('supplyDownstreamEnabled', True) is not False: for other_pid, other_cfg in PRODUCT_CONFIGS.items(): other_inputs = other_cfg.get('inputs', []) if any(inp['id'] == pid for inp in other_inputs): other_state = state.get('products', {}).get(other_pid) if other_state and other_state.get('unlocked') and other_state.get('lines', 0) > 0: inp_match = next(inp for inp in other_inputs if inp['id'] == pid) reserved_stock = max(reserved_stock, float(inp_match['amount'] * other_state['lines'])) avail = max(0.0, p_state.get('stock', 0.0) - reserved_stock) if avail > 0: max_sell = logistics * 0.75 * dt sold = min(avail, max_sell) p_state['stock'] -= sold earned = sold * p_cfg['basePrice'] * logistics_price_mult state['cash'] = state.get('cash', 0.0) + earned state['lifetimeRevenue'] = state.get('lifetimeRevenue', 0.0) + earned # Update valuation facility_assets = sum( p_state.get('lines', 0) * PRODUCT_CONFIGS[pid]['baseFacilityCost'] for pid, p_state in state.get('products', {}).items() if pid in PRODUCT_CONFIGS ) state['valuation'] = state.get('cash', 0.0) + facility_assets + (state.get('lifetimeRevenue', 0.0) * 0.15) return state def apply_actions(state: dict, actions: list) -> dict: """Validates and executes queued actions sent from the frontend.""" for act in actions: atype = act.get('type') if atype == 'UNLOCK_PRODUCT': pid = act.get('productId') if pid in PRODUCT_CONFIGS and pid in state.get('products', {}): p_cfg = PRODUCT_CONFIGS[pid] cost = p_cfg['unlockCost'] if state.get('cash', 0.0) >= cost and not state['products'][pid].get('unlocked'): state['cash'] -= cost state['products'][pid]['unlocked'] = True state['products'][pid]['lines'] = 1 elif atype == 'BUY_LINES': pid = act.get('productId') if pid in PRODUCT_CONFIGS and pid in state.get('products', {}): p_cfg = PRODUCT_CONFIGS[pid] current_lines = state['products'][pid].get('lines', 0) client_rep = state.get('repAllocations', {}).get('clients', 0) if isinstance(state.get('repAllocations'), dict) else 0 line_discount = min(0.50, 0.002 * client_rep) cost = p_cfg['baseFacilityCost'] * math.pow(1.15, current_lines) * (1.0 - line_discount) if state.get('cash', 0.0) >= cost: state['cash'] -= cost state['products'][pid]['lines'] = current_lines + 1 elif atype == 'HIRE_MANAGER': pid = act.get('productId') if pid in PRODUCT_CONFIGS and pid in state.get('products', {}): cost = PRODUCT_CONFIGS[pid]['baseManagerCost'] if state.get('cash', 0.0) >= cost and not state['products'][pid].get('manager'): state['cash'] -= cost state['products'][pid]['manager'] = True state['stats']['managersHired'] = state['stats'].get('managersHired', 0) + 1 elif atype == 'BUY_LOGISTICS': pid = act.get('productId') if pid in PRODUCT_CONFIGS and pid in state.get('products', {}): current_logistics = state['products'][pid].get('logistics', 0) cost = PRODUCT_CONFIGS[pid]['baseLogisticsCost'] * math.pow(1.15, current_logistics) if state.get('cash', 0.0) >= cost: state['cash'] -= cost state['products'][pid]['logistics'] = current_logistics + 1 elif atype == 'TOGGLE_LOGISTICS': pid = act.get('productId') if pid in state.get('products', {}): current = state['products'][pid].get('logisticsEnabled', True) is not False state['products'][pid]['logisticsEnabled'] = not current elif atype == 'TOGGLE_DOWNSTREAM_SUPPLY': pid = act.get('productId') if pid in state.get('products', {}): current = state['products'][pid].get('supplyDownstreamEnabled', True) is not False state['products'][pid]['supplyDownstreamEnabled'] = not current elif atype == 'SET_PRODUCTION_TARGET': pid = act.get('productId') target = act.get('target', 1.0) if pid in state.get('products', {}): try: target_val = float(target) if 0.0 <= target_val <= 1.0: state['products'][pid]['productionTarget'] = target_val except (ValueError, TypeError): pass elif atype == 'TRIGGER_IPO': earned = calculate_shares_earned(state.get('lifetimeRevenue', 0.0)) if earned > 0: preserved_shares = state.get('ventureShares', 0) + earned preserved_total_shares = state.get('totalVentureShares', 0) + earned preserved_perks = state.get('boardroomPerks', {}) preserved_ipos = state.get('ipoCount', 0) + 1 # Calculate preserved reputation from brand_legacy perk (10% to 50%) brand_lvl = preserved_perks.get('brand_legacy', 0) rep_retention_ratios = [0.0, 0.10, 0.20, 0.30, 0.40, 0.50] ratio = rep_retention_ratios[min(brand_lvl, len(rep_retention_ratios) - 1)] total_current_rep = state.get('reputation', 0) allocs = state.get('repAllocations', {}) total_rep_accumulated = total_current_rep + sum(allocs.values()) if isinstance(allocs, dict) else total_current_rep preserved_rep = int(math.floor(total_rep_accumulated * ratio)) # Starting cash from angel_syndicate angel_lvl = preserved_perks.get('angel_syndicate', 0) starting_cash_bonuses = [30.0, 500.0, 5000.0, 50000.0, 500000.0, 5000000.0] new_starting_cash = starting_cash_bonuses[min(angel_lvl, len(starting_cash_bonuses) - 1)] # Reset to clean initial state state = get_initial_state() state['cash'] = new_starting_cash state['ventureShares'] = preserved_shares state['totalVentureShares'] = preserved_total_shares state['boardroomPerks'] = preserved_perks state['ipoCount'] = preserved_ipos state['reputation'] = preserved_rep state['repAllocations'] = {'b2b': 0, 'clients': 0, 'brand': 0} elif atype == 'BUY_PERK': perk_id = act.get('perkId') if perk_id in BOARDROOM_PERK_COSTS: current_lvl = state.get('boardroomPerks', {}).get(perk_id, 0) costs = BOARDROOM_PERK_COSTS[perk_id] if current_lvl < len(costs): req_shares = costs[current_lvl] if state.get('ventureShares', 0) >= req_shares: state['ventureShares'] -= req_shares state['boardroomPerks'][perk_id] = current_lvl + 1 elif atype == 'ALLOCATE_REP': dept = act.get('department') delta = act.get('delta', 0) if 'repAllocations' not in state or not isinstance(state['repAllocations'], dict): state['repAllocations'] = {'b2b': 0, 'clients': 0, 'brand': 0} allocs = state['repAllocations'] if delta == 'reset': # Return all points to available pool freed = sum(allocs.values()) state['reputation'] = state.get('reputation', 0) + freed state['repAllocations'] = {'b2b': 0, 'clients': 0, 'brand': 0} elif dept in ('b2b', 'clients', 'brand'): try: delta_int = int(delta) if delta_int > 0: # Allocating points from pool to_alloc = min(delta_int, state.get('reputation', 0)) if to_alloc > 0: state['reputation'] -= to_alloc allocs[dept] = allocs.get(dept, 0) + to_alloc elif delta_int < 0: # De-allocating points back to pool to_free = min(abs(delta_int), allocs.get(dept, 0)) if to_free > 0: allocs[dept] -= to_free state['reputation'] = state.get('reputation', 0) + to_free except (ValueError, TypeError): pass return state def sanitize_state(raw: dict) -> dict: """Ensures incoming state doesn't have NaN or negative numbers and dynamically populates new products.""" if not isinstance(raw, dict): return get_initial_state() if 'repAllocations' not in raw or not isinstance(raw['repAllocations'], dict): raw['repAllocations'] = {'b2b': 0, 'clients': 0, 'brand': 0} else: for k in ('b2b', 'clients', 'brand'): if k not in raw['repAllocations'] or not isinstance(raw['repAllocations'][k], (int, float)): raw['repAllocations'][k] = 0 else: raw['repAllocations'][k] = max(0, int(raw['repAllocations'][k])) # Dynamic re-hydration: Ensure any new products added to the catalog exist in the player's saved state if 'products' not in raw or not isinstance(raw['products'], dict): raw['products'] = {} for pid, p in PRODUCT_CONFIGS.items(): if pid not in raw['products']: is_unlocked = p.get('unlockedDefault', False) raw['products'][pid] = { 'unlocked': is_unlocked, 'stock': 0.0, 'lines': 1 if is_unlocked else 0, 'manager': False, 'logistics': 1 if is_unlocked else 0, 'logisticsEnabled': True, 'supplyDownstreamEnabled': True, 'productionTarget': 1.0, 'cycleProgress': 0.0, 'isBottlenecked': False, 'totalProduced': 0.0 } return raw