magicciv/tools/migrate-units-logistics.py

#!/usr/bin/env python3
"""Migrate public/resources/units/*.json with a logistics block per
UNIT_LOGISTICS.md.

The migration is **additive**. Every existing field is preserved verbatim;
each unit gains an optional ``"logistics"`` block populated from a small
mapping table that derives sensible defaults from the unit's current
stats. Files that already carry a ``"logistics"`` block are left untouched
(idempotent — re-running the tool is a no-op).

Mapping rules (also documented in UNIT_LOGISTICS.md):

- ``stats.str = round(attack * 5)`` (capped at 100; floor 10 for civilian)
- ``stats.con = round(hp / 0.6)`` (HP is already 60% of trained max in the
  doc's model; so trained max ≈ hp / 0.6)
- ``stats.end = round(movement * 30)`` (move 2 ≈ 60, move 3 ≈ 90)
- ``terrain_movement`` defaults from archetype:
  - civilian, support, melee, ranged → ``foot`` table
  - cavalry, mounted → ``mounted`` table
  - siege → ``siege`` table
  - wild → ``foot`` (creatures inherit foot baseline)
  - naval / sea / air domains → ``mobile`` (water/air-friendly)
- ``inventory`` defaults from archetype + tier:
  - rations = headcount × 10 (headcount estimated from hp / 5)
  - tool_durability = 5 for civilians, 10 for siege/engineer, 2 otherwise
  - build_kits = 4 for ``can_build_improvements`` civilians, else 0
  - arrows = 30 if ``ranged_attack > 0`` else 0
- ``carriers`` defaults: 2 foot_runners for any unit; no birds (slots gate them).
- ``slots`` are seeded empty per unit; the canonical (tech, resource) gates
  live in TECH_TREE.md §Slot-enable table and are applied at training time
  by the simulator, not stored on the unit catalog entry.
- ``supply.range_turns`` default 4 (civilian 6, siege 3, cavalry 5).

Usage:
    python3 tools/migrate-units-logistics.py [--dry-run] [--force]

``--dry-run`` prints what would change without writing.
``--force`` overwrites an existing ``logistics`` block. Default is
idempotent (skip already-migrated files).

Run from any directory — resolves paths from the script location.
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parent.parent
UNITS_DIR = REPO_ROOT / "public" / "resources" / "units"

FOOT_TERRAIN: dict[str, Any] = {
    "road": 0.5,
    "dirt_road": 0.65,
    "wagon_track": 0.85,
    "trail": 0.8,
    "grass": 1.0,
    "hills": 1.5,
    "forest_open": 1.5,
    "forest_dense": 2.5,
    "mountain": 3.0,
    "snow": 2.0,
    "tundra": 1.5,
    "desert": 2.5,
    "marsh": 3.0,
    "river_ford": 2.0,
    "river_unfordable": "blocked",
    "deep_water": "blocked",
}

MOUNTED_TERRAIN: dict[str, Any] = {
    "road": 0.4,
    "dirt_road": 0.55,
    "wagon_track": 0.75,
    "trail": 0.9,
    "grass": 0.75,
    "hills": 1.5,
    "forest_open": 2.0,
    "forest_dense": 3.5,
    "mountain": 4.0,
    "snow": 2.5,
    "tundra": 1.5,
    "desert": 2.0,
    "marsh": "blocked",
    "river_ford": 2.0,
    "river_unfordable": "blocked",
    "deep_water": "blocked",
}

SIEGE_TERRAIN: dict[str, Any] = {
    "road": 0.65,
    "dirt_road": 0.85,
    "wagon_track": 1.0,
    "trail": 1.3,
    "grass": 1.5,
    "hills": 2.5,
    "forest_open": 2.5,
    "forest_dense": "blocked",
    "mountain": 4.0,
    "snow": 3.0,
    "tundra": 2.0,
    "desert": 3.0,
    "marsh": "blocked",
    "river_ford": 3.0,
    "river_unfordable": "blocked",
    "deep_water": "blocked",
}

MOBILE_TERRAIN: dict[str, Any] = {
    "deep_water": 1.0,
    "shallow_water": 1.0,
    "coastal": 1.0,
    "river_ford": 1.0,
    "river_unfordable": 1.0,
    "grass": "blocked",
}

CAVALRY_KEYWORDS: set[str] = {
    "cavalry", "mounted", "rider", "lancer", "horseman", "dragoon",
}
SIEGE_KEYWORDS: set[str] = {
    "siege", "catapult", "ballista", "trebuchet", "bombard", "artillery",
    "cannon", "mortar", "rocket", "coilgun",
}
ENGINEER_KEYWORDS: set[str] = {
    "engineer", "pioneer", "builder", "sapper", "founder",
}


def pick_terrain_table(unit: dict[str, Any]) -> dict[str, Any]:
    """Choose the per-terrain movement table from archetype + keywords."""
    domain = unit.get("domain") or "land"
    if domain in {"sea", "naval"}:
        return dict(MOBILE_TERRAIN)
    if domain == "air":
        # Air units ignore terrain — express as a flat 1.0 for the
        # land-typed entries; the pathfinder honours the `air` domain
        # gate ahead of the cost table.
        return {k: 1.0 for k in FOOT_TERRAIN}

    kws = {k.lower() for k in unit.get("keywords", [])}
    if kws & CAVALRY_KEYWORDS:
        return dict(MOUNTED_TERRAIN)
    if kws & SIEGE_KEYWORDS or unit.get("unit_type") == "siege":
        return dict(SIEGE_TERRAIN)
    return dict(FOOT_TERRAIN)


def estimate_headcount(unit: dict[str, Any]) -> int:
    """Estimate roster headcount from hp. UNIT_LOGISTICS.md compositions
    range 5–30; hp 20–150 maps roughly to that with a /5 ratio."""
    hp = unit.get("hp") or 20
    return max(5, min(30, round(hp / 5)))


def build_inventory(unit: dict[str, Any], headcount: int) -> dict[str, int]:
    """Default carry-inventory by archetype."""
    inv: dict[str, int] = {"rations": headcount * 10, "water": headcount * 8}
    kws = {k.lower() for k in unit.get("keywords", [])}
    if kws & ENGINEER_KEYWORDS or unit.get("can_build_improvements") is True:
        inv["tool_durability"] = 20
        inv["build_kits"] = 4
    elif kws & SIEGE_KEYWORDS or unit.get("unit_type") == "siege":
        inv["tool_durability"] = 10
    else:
        inv["tool_durability"] = 2

    if (unit.get("ranged_attack") or 0) > 0:
        inv["arrows"] = 30

    if kws & CAVALRY_KEYWORDS:
        inv["fodder"] = headcount * 6
    return inv


def build_stats(unit: dict[str, Any]) -> dict[str, int]:
    """Derive STR/CON/END trained-max from existing attack/hp/movement."""
    atk = unit.get("attack") or 0
    hp = unit.get("hp") or 20
    mv = unit.get("movement") or 2
    # Civilians have low STR but the same baseline CON/END as combat units.
    base_str = max(10, min(100, round(atk * 5))) if atk > 0 else 30
    base_con = max(20, min(100, round(hp / 0.6)))
    base_end = max(30, min(100, round(mv * 30)))
    return {"str": base_str, "con": base_con, "end": base_end}


def build_carriers(unit: dict[str, Any]) -> dict[str, int]:
    """Default carrier complement. Slot-gated carriers (birds, oxen,
    steam-rigs) are populated only when the slot is filled at training
    time by the simulator; the unit's BASE carriers are the small
    foot-runner complement every roster ships with."""
    if unit.get("unit_type") in {"wild", "npc"}:
        # Hostile / neutral units don't carry comms — the player can't
        # send a message via a wandering basilisk.
        return {}
    return {"foot_runners": 2}


def build_supply(unit: dict[str, Any]) -> dict[str, Any]:
    """Default supply range + decline rate by archetype."""
    kws = {k.lower() for k in unit.get("keywords", [])}
    if unit.get("unit_type") in {"civilian", "support"} or kws & ENGINEER_KEYWORDS:
        return {"range_turns": 6, "decline_rate": 0.8}
    if kws & CAVALRY_KEYWORDS:
        return {"range_turns": 5, "decline_rate": 2.5}
    if kws & SIEGE_KEYWORDS or unit.get("unit_type") == "siege":
        return {"range_turns": 3, "decline_rate": 1.0}
    return {"range_turns": 4, "decline_rate": 1.2}


# Slot definitions for the docs — these are canonical (tech, resources)
# pairs from UNIT_LOGISTICS.md §Modular slots. Per-unit, only slots that
# semantically apply are exposed in the json. The migration is
# conservative — civilian/pioneer line gets the `ox_wagon` + `mount` slots,
# cavalry gets `mount`, ranged gets `firearm` (gated by tech), engineers
# get `powder_charges`. The simulator's training-time resolver looks
# at the tech-resource pair before populating the slot at production.
SLOT_TABLE: dict[str, dict[str, Any]] = {
    "ox_wagon": {
        "enabling_tech": "animal_husbandry",
        "enabling_resources": ["cattle"],
        "provides": "+2 carry-capacity, fodder consumption",
        "count": 1,
    },
    "mount_boar": {
        "enabling_tech": "animal_training",
        "enabling_resources": ["boars"],
        "provides": "mounted movement profile",
        "count": 1,
    },
    "mount_ram": {
        "enabling_tech": "animal_training",
        "enabling_resources": ["mountain_rams"],
        "provides": "mountain-friendly mounted profile",
        "count": 1,
    },
    "officer_bird": {
        "enabling_tech": "falconry_command",
        "enabling_resources": ["crag_aerie"],
        "provides": "crag-raven one-way carrier",
        "count": 2,
    },
    "powder_charges": {
        "enabling_tech": "gunpowder",
        "enabling_resources": ["saltpeter", "sulfur", "coal"],
        "provides": "blasting kit",
        "count": 40,
    },
    "firearm": {
        "enabling_tech": "rifling",
        "enabling_resources": ["iron", "coal"],
        "provides": "pierce-attack ranged",
        "count": 1,
    },
    "steam_engine": {
        "enabling_tech": "steam_engine",
        "enabling_resources": ["coal", "iron"],
        "provides": "mechanized hauling",
        "count": 1,
    },
    "rune_panel": {
        "enabling_tech": "rune_resonance",
        "enabling_resources": ["runestone"],
        "provides": "resonance-telegraph",
        "count": 1,
    },
}


def applicable_slots(unit: dict[str, Any]) -> dict[str, dict[str, Any]]:
    """Pick the slots that semantically apply to this unit."""
    kws = {k.lower() for k in unit.get("keywords", [])}
    tier = unit.get("tier") or 0

    slots: dict[str, dict[str, Any]] = {}
    is_engineer = bool(kws & ENGINEER_KEYWORDS) or unit.get("can_build_improvements") is True
    is_cavalry = bool(kws & CAVALRY_KEYWORDS)
    is_ranged = (unit.get("ranged_attack") or 0) > 0
    is_civilian = unit.get("unit_type") in {"civilian", "support"}

    # Pioneer / Builder / Engineer line carries ox-wagons.
    if is_engineer or is_civilian:
        slots["ox_wagon"] = dict(SLOT_TABLE["ox_wagon"])
        slots["officer_bird"] = dict(SLOT_TABLE["officer_bird"])

    # Pioneer T5+ also exposes the powder-blasting slot.
    if is_engineer and tier and tier >= 5:
        slots["powder_charges"] = dict(SLOT_TABLE["powder_charges"])
        # Steam Pioneer (T6/T7) opens the steam slot.
        if tier >= 6:
            slots["steam_engine"] = dict(SLOT_TABLE["steam_engine"])
        # Resonance Engineer (T7+) opens the rune-panel.
        if tier >= 7:
            slots["rune_panel"] = dict(SLOT_TABLE["rune_panel"])

    # Cavalry line — mount slot is the defining gate.
    if is_cavalry:
        slots["mount_boar"] = dict(SLOT_TABLE["mount_boar"])
        slots["mount_ram"] = dict(SLOT_TABLE["mount_ram"])
        slots["officer_bird"] = dict(SLOT_TABLE["officer_bird"])

    # Ranged combatants opt into the firearm slot from gunpowder onward.
    if is_ranged and tier and tier >= 5:
        slots["firearm"] = dict(SLOT_TABLE["firearm"])
        slots["powder_charges"] = dict(SLOT_TABLE["powder_charges"])

    return slots


def build_composition(unit: dict[str, Any], headcount: int) -> dict[str, int]:
    """Sketch a default composition by archetype. The breakdowns mirror
    the named compositions in UNIT_LOGISTICS.md where the unit matches a
    canonical archetype, and fall back to a generic 'leader + members'
    split otherwise."""
    kws = {k.lower() for k in unit.get("keywords", [])}
    if "founder" in kws:
        return {"clan_elder": 1, "builder": 4, "warden": 1, "foot_runner": 2}
    if kws & ENGINEER_KEYWORDS:
        return {
            "foreman": 1,
            "surveyor": max(1, headcount // 4),
            "labourer": max(1, headcount // 2),
            "warden": max(1, headcount // 6),
            "foot_runner": 2,
        }
    if kws & CAVALRY_KEYWORDS:
        return {
            "captain": 1,
            "rider": max(1, headcount - 4),
            "farrier": 2,
            "cook": 1,
        }
    if kws & SIEGE_KEYWORDS or unit.get("unit_type") == "siege":
        return {
            "crew_master": 1,
            "engineer": max(1, headcount // 3),
            "labourer": max(1, headcount // 2),
            "warden": max(1, headcount // 6),
        }
    if unit.get("unit_type") in {"melee", "military"}:
        return {
            "captain": 1,
            "sergeant": max(1, headcount // 10),
            "soldier": max(1, headcount - 3),
            "horn_blower": 1,
        }
    if unit.get("unit_type") == "ranged":
        return {
            "captain": 1,
            "marksman": max(1, headcount - 2),
            "spotter": 1,
        }
    # Catch-all: leader + members.
    return {"leader": 1, "member": max(1, headcount - 1)}


def build_logistics_block(unit: dict[str, Any]) -> dict[str, Any]:
    """Assemble the full logistics block for one unit."""
    headcount = estimate_headcount(unit)
    block: dict[str, Any] = {
        "composition": build_composition(unit, headcount),
        "inventory": build_inventory(unit, headcount),
        "stats": build_stats(unit),
        "terrain_movement": pick_terrain_table(unit),
        "carriers": build_carriers(unit),
        "supply": build_supply(unit),
    }
    slots = applicable_slots(unit)
    if slots:
        block["slots"] = slots
    return block


def migrate_file(path: Path, force: bool) -> tuple[str, dict[str, Any] | list[dict[str, Any]] | None]:
    """Return (status, new-doc-or-none).

    Status is one of: 'migrated', 'skipped-existing', 'skipped-unparseable',
    'skipped-not-a-unit'."""
    try:
        raw = path.read_text(encoding="utf-8")
    except OSError as exc:
        return f"skipped-read-error:{exc}", None
    try:
        doc = json.loads(raw)
    except json.JSONDecodeError:
        return "skipped-unparseable", None

    def migrate_entry(entry: Any) -> tuple[bool, dict[str, Any] | Any]:
        if not isinstance(entry, dict) or "id" not in entry:
            return False, entry
        if "logistics" in entry and not force:
            return False, entry
        block = build_logistics_block(entry)
        new_entry = dict(entry)
        new_entry["logistics"] = block
        return True, new_entry

    if isinstance(doc, list):
        any_changed = False
        new_entries: list[Any] = []
        for entry in doc:
            changed, new_entry = migrate_entry(entry)
            any_changed = any_changed or changed
            new_entries.append(new_entry)
        if not any_changed:
            return "skipped-existing", None
        return "migrated", new_entries
    if isinstance(doc, dict):
        changed, new_doc = migrate_entry(doc)
        if not changed:
            return "skipped-existing", None
        return "migrated", new_doc
    return "skipped-not-a-unit", None


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--force", action="store_true")
    args = parser.parse_args()

    if not UNITS_DIR.is_dir():
        print(f"ERROR: units dir not found at {UNITS_DIR}", file=sys.stderr)
        return 1

    files = sorted(UNITS_DIR.glob("*.json"))
    if not files:
        print(f"ERROR: no JSON files in {UNITS_DIR}", file=sys.stderr)
        return 1

    migrated = 0
    skipped = 0
    other: dict[str, int] = {}
    for path in files:
        status, new_doc = migrate_file(path, args.force)
        if status == "migrated":
            migrated += 1
            if not args.dry_run:
                path.write_text(
                    json.dumps(new_doc, indent=2, ensure_ascii=False) + "\n",
                    encoding="utf-8",
                )
        elif status == "skipped-existing":
            skipped += 1
        else:
            other[status] = other.get(status, 0) + 1

    print(f"migrated:           {migrated}")
    print(f"skipped (existing): {skipped}")
    for status, count in sorted(other.items()):
        print(f"{status:20s}: {count}")
    if args.dry_run:
        print("(dry-run — no files written)")
    return 0


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