dota_factory/tools/threat_report.py

#!/usr/bin/env python3
"""Report threat values, ship costs, ratios, and belt feasibility.

Reads recipes.toml, ships.toml, modules.toml, and world.toml and prints:

  1. Item threats — production-seconds per item unit, resolved
     recursively over the recipe tree (REQ-THREAT-ITEM):
       - non-reprocessing recipe: (duration + sum(input threat * qty))
         divided by the output amount, so threat is per unit;
       - multiple recipes: the maximum across them;
       - recipes consuming scrap participate only if no scrap-free
         recipe produces the item (fallback rule);
       - reprocessing-only items: (scrap threat * scrap per cycle
         + duration) / normalized probability (REQ-THREAT-ITEM).
     Scrap threat is the constant 1 / world.scrap_per_threat
     (REQ-THREAT-SCRAP).
  2. Module contributions — material threat + production time per
     module instance (the amount a module adds to a ship's threat).
  3. Ship threats — hull-only and fitted with default_modules (the
     loadout enemy waves spawn with, REQ-WAV-DEFAULT-MODULES).
  4. Producer:consumer ratios — buildings of the input recipe needed
     per building of the consuming recipe at 100% throughput.
  5. Belt feasibility — input demand in items/s per building vs. the
     single-belt cap (belt_speed_mps / tile_size_m, in items/s).

NOTE: the semantics described above — per-unit division, scrap fallback
rule, fixpoint resolution including reprocessing, and the commit-all-
recipes max rule — are implemented in both this script and in
src/lib/sim/ThreatCostCalculator.cpp. The two should produce identical
values for any given config.

Usage (from the repository root or anywhere else):

    python dota_factory/tools/threat_report.py
    python dota_factory/tools/threat_report.py --config-dir path/to/config

By default the config directory is resolved relative to this script
(../bin/app/data/config). Requires the 'toml' package on Python < 3.11
(pip install --user toml); on 3.11+ the standard tomllib is used.
"""

import argparse
import os
import sys


def load_toml(path):
    try:
        import tomllib
        with open(path, "rb") as fh:
            return tomllib.load(fh)
    except ImportError:
        import toml
        return toml.load(path)


def consumes_scrap(recipe):
    return any(inp["item"] == "scrap" for inp in recipe.get("inputs", []))


def recipe_threat_per_unit(recipe, output, item_threat):
    threat = recipe["duration_seconds"]
    for inp in recipe.get("inputs", []):
        if inp["item"] not in item_threat:
            return None
        threat += item_threat[inp["item"]] * inp["amount"]
    return threat / output["amount"]


def resolve_items(recipes, scrap_threat):
    """Return {item: threat} resolved per REQ-THREAT-ITEM."""
    non_repro = [r for r in recipes if r["building"] != "reprocessing_plant"]
    repro = [r for r in recipes if r["building"] == "reprocessing_plant"]

    # Items with at least one scrap-free producer: their scrap-consuming
    # recipes never participate (fallback rule).
    scrap_free_items = set()
    for recipe in non_repro:
        if not consumes_scrap(recipe):
            for output in recipe.get("outputs", []):
                scrap_free_items.add(output["item"])

    def eligible(recipe, output):
        if consumes_scrap(recipe) and output["item"] in scrap_free_items:
            return False
        return True

    item_threat = {"scrap": scrap_threat}

    # All eligible (recipe, output) pairs per item: the max rule requires
    # committing an item only once EVERY eligible recipe for it is
    # computable — otherwise a shallow shortcut recipe that resolves one
    # pass earlier than the base path would win and underprice the item.
    recipes_per_item = {}
    for recipe in non_repro:
        for output in recipe.get("outputs", []):
            if eligible(recipe, output):
                recipes_per_item.setdefault(output["item"], []).append(
                    (recipe, output))

    def resolve_pass(require_all_recipes=True):
        progress = False
        best = {}
        for item, pairs in recipes_per_item.items():
            if item in item_threat:
                continue
            threats = [recipe_threat_per_unit(r, o, item_threat)
                       for r, o in pairs]
            if require_all_recipes and any(t is None for t in threats):
                continue
            threats = [t for t in threats if t is not None]
            if threats:
                best[item] = max(threats)
        for item, threat in best.items():
            item_threat[item] = threat
            progress = True
        return progress

    def repro_pass():
        progress = False
        for recipe in repro:
            scrap_per_cycle = sum(inp["amount"]
                                  for inp in recipe.get("inputs", []))
            total_weight = sum(out.get("probability", 1.0)
                               for out in recipe.get("outputs", []))
            for output in recipe.get("outputs", []):
                # Reprocessing defines an item's threat only when nothing
                # else produces it (REQ-THREAT-ITEM).
                if output["item"] in item_threat:
                    continue
                if output["item"] in scrap_free_items:
                    continue
                probability = output.get("probability", 1.0) / total_weight
                if probability <= 0.0:
                    continue
                item_threat[output["item"]] = (
                    (scrap_threat * scrap_per_cycle
                     + recipe["duration_seconds"]) / probability)
                progress = True
        return progress

    # Iterate to a fixpoint: items downstream of reprocessing-only items
    # (e.g. capital parts built from voidsteel) need another recipe pass
    # after the reprocessing pass has resolved their inputs.
    progress = True
    while progress:
        progress = resolve_pass()
        progress = repro_pass() or progress

    # Deadlock guard: if recipe cycles keep items waiting on each other,
    # fall back to committing with the computable subset of recipes.
    progress = True
    while progress:
        progress = resolve_pass(require_all_recipes=False)
        progress = repro_pass() or progress

    return item_threat


def material_threat(materials, item_threat, missing):
    total = 0.0
    for material in materials:
        if material["item"] in item_threat:
            total += item_threat[material["item"]] * material["amount"]
        else:
            missing.add(material["item"])
    return total


def main():
    default_dir = os.path.normpath(os.path.join(
        os.path.dirname(os.path.abspath(__file__)),
        "..", "bin", "app", "data", "config"))
    parser = argparse.ArgumentParser(
        description="Report threat values, ship costs, ratios, and belt"
                    " feasibility from the config files.")
    parser.add_argument("--config-dir", default=default_dir,
                        help="directory containing the config toml files"
                             " (default: %(default)s)")
    args = parser.parse_args()

    recipes = load_toml(os.path.join(args.config_dir, "recipes.toml"))["recipe"]
    ships = load_toml(os.path.join(args.config_dir, "ships.toml"))["ship"]
    modules = load_toml(os.path.join(args.config_dir, "modules.toml"))["module"]
    world = load_toml(os.path.join(args.config_dir, "world.toml"))["world"]

    scrap_per_threat = world.get("scrap_per_threat", 0.0)
    scrap_threat = 1.0 / scrap_per_threat if scrap_per_threat > 0 else 0.0
    belt_cap = world["belt_speed_mps"] / world["tile_size_m"]  # items/s

    item_threat = resolve_items(recipes, scrap_threat)
    missing = set()

    print("scrap_per_threat = {}  =>  threat(scrap) = {:.2f}".format(
        scrap_per_threat, scrap_threat))
    print()

    print("== item threats (production-seconds per unit) ==")
    for item in sorted(item_threat):
        if item != "scrap":
            print("  {:24s} {:10.2f}".format(item, item_threat[item]))

    print()
    print("== module contributions (materials + production time) ==")
    module_contribution = {}
    for module in modules:
        contribution = (material_threat(module["materials"], item_threat, missing)
                        + module["production_time_seconds"])
        module_contribution[module["id"]] = contribution
        print("  {:24s} {:10.2f}".format(module["id"], contribution))

    print()
    print("== ship threats (hull-only / fitted with default_modules) ==")
    for ship in ships:
        hull = (material_threat(ship["schematic"]["materials"], item_threat, missing)
                + ship["schematic"]["production_time_seconds"])
        fitted = hull
        for placed in ship.get("default_modules", []):
            fitted += module_contribution.get(placed["type"], 0.0)
        print("  {:16s} hull {:9.2f}   fitted {:9.2f}".format(
            ship["id"], hull, fitted))

    print()
    print("== producer:consumer ratios (producing buildings per consuming"
          " building) ==")
    producers = {}  # item -> [(recipe id, items/s per building)]
    for recipe in recipes:
        if recipe["building"] == "reprocessing_plant":
            continue
        for output in recipe.get("outputs", []):
            rate = output["amount"] / recipe["duration_seconds"]
            producers.setdefault(output["item"], []).append((recipe["id"], rate))
    for recipe in recipes:
        for inp in recipe.get("inputs", []):
            need = inp["amount"] / recipe["duration_seconds"]
            for producer_id, supply in producers.get(inp["item"], []):
                print("  {:20s} -> {:24s} {:6.3f}".format(
                    producer_id, recipe["id"], need / supply))

    print()
    print("== belt feasibility (input demand items/s per building,"
          " cap {:.2f}/s) ==".format(belt_cap))
    over = False
    for recipe in recipes:
        for inp in recipe.get("inputs", []):
            rate = inp["amount"] / recipe["duration_seconds"]
            if rate > belt_cap:
                print("  OVER: {} <- {}: {:.2f}/s".format(
                    recipe["id"], inp["item"], rate))
                over = True
    if not over:
        print("  all input demands under the single-belt cap")

    if missing:
        print()
        for item in sorted(missing):
            print("WARNING: no threat value for material '{}'".format(item))

    return 0


if __name__ == "__main__":
    sys.exit(main())