scripts/tune_from_annotations.py

from __future__ import annotations

import argparse
import json
import re
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Literal, Mapping

import numpy as np

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from life_game.game import GAME_MODES, SANDBOX_MODE, new_game
from life_game.tuning import (
    ALLOWED_DATA_KEYS,
    DATA_RANGES,
    ModeTuning,
    TUNING_SCHEMA,
    TuningProfile,
    parse_tuning_profile,
    tuning_profile_to_dict,
)

Direction = Literal["easier", "harder", "skip"]

EASIER_PATTERNS: tuple[tuple[str, int], ...] = (
    ("far too hard", 2),
    ("very hard", 2),
    ("too hard initially", 2),
    ("did not survive", 2),
    ("too many enemies", 1),
    ("too hard", 1),
    ("to hard", 1),
    ("bit hard", 1),
    ("cannot interact", 1),
)
HARDER_PATTERNS: tuple[tuple[str, int], ...] = (
    ("far too easy", 2),
    ("impossible to get killed", 2),
    ("have to do nothing", 2),
    ("make it a bit harder", 1),
    ("needs to be a bit harder", 1),
    ("bit harder", 1),
    ("too easy", 1),
    ("bit too easy", 1),
    ("bit easy", 1),
    ("difficulty is a bit low", 1),
    ("right now a bit easy", 1),
)
SKIP_PATTERNS = (
    "drop this game",
    "drop",
    "keep as is",
    "well calibrated",
    "difficulty is good",
)
SEMANTIC_ONLY_PATTERNS = (
    "extension",
    "enemy types",
    "shot types",
    "different directions",
    "does not end",
    "phase",
    "visible",
    "visual",
)
TARGET_KEYS = ("target_score", "target_waves", "target", "target_claimed", "target_discharges", "target_allies")


@dataclass(frozen=True)
class AnnotationDecision:
    mode: str
    direction: Direction
    severity: int
    reason: str


def main() -> None:
    parser = argparse.ArgumentParser(description="Generate deterministic tuning from gameplay annotation JSONL.")
    parser.add_argument("--annotations", default="annotations/game_feedback.jsonl")
    parser.add_argument("--output", default="profiles/annotation_tuning.json")
    parser.add_argument("--size", type=int, default=24)
    parser.add_argument("--report", action="store_true")
    args = parser.parse_args()

    records = _load_latest_annotations(Path(args.annotations))
    profile, decisions = tune_from_annotations(records, max(12, int(args.size)))

    output = Path(args.output).expanduser()
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(tuning_profile_to_dict(profile), indent=2, sort_keys=True) + "\n", encoding="utf-8")
    print(f"Wrote {output}")

    if args.report:
        for decision in decisions:
            print(f"{decision.mode}: {decision.direction} severity={decision.severity} - {decision.reason}")


def tune_from_annotations(records: Mapping[str, Mapping[str, Any]], size: int = 24) -> tuple[TuningProfile, tuple[AnnotationDecision, ...]]:
    modes: dict[str, ModeTuning] = {}
    decisions: list[AnnotationDecision] = []
    for index, mode in enumerate(mode for mode in GAME_MODES if mode != SANDBOX_MODE):
        record = records.get(mode)
        if record is None:
            decisions.append(AnnotationDecision(mode, "skip", 0, "no annotation"))
            continue
        decision = decide_annotation(record)
        decisions.append(decision)
        if decision.direction == "skip":
            continue

        game = new_game(size, mode, np.random.default_rng(index + 301))
        tuning = _build_mode_tuning(game.health, game.max_health, dict(game.data), decision.direction, decision.severity)
        if tuning.health is not None or tuning.max_health is not None or tuning.data:
            modes[mode] = tuning

    profile = TuningProfile(
        modes=modes,
        description=(
            "Generated deterministically from annotations/game_feedback.jsonl. "
            "Only health and existing target counters are tuned; semantic notes remain code-change candidates."
        ),
    )
    parse_tuning_profile(tuning_profile_to_dict(profile))
    return profile, tuple(decisions)


def decide_annotation(record: Mapping[str, Any]) -> AnnotationDecision:
    mode = str(record.get("mode", ""))
    text = _normalized_feedback(record)
    if not text:
        return AnnotationDecision(mode, "skip", 0, "empty feedback")

    if any(pattern in text for pattern in SKIP_PATTERNS):
        if _contains_explicit_direction(text):
            # Explicit difficulty feedback wins over broad extension prose, but not explicit drops.
            if "drop" in text:
                return AnnotationDecision(mode, "skip", 0, "drop annotation")
        else:
            return AnnotationDecision(mode, "skip", 0, "annotation says keep/drop/no difficulty change")

    easier = _best_match(text, EASIER_PATTERNS)
    harder = _best_match(text, HARDER_PATTERNS)
    if harder[1] > easier[1] or (harder[1] > 0 and harder[1] == easier[1] and len(harder[0]) > len(easier[0])):
        return AnnotationDecision(mode, "harder", _effective_severity(record, harder[1], "harder"), harder[0])
    if easier[1] > 0:
        return AnnotationDecision(mode, "easier", _effective_severity(record, easier[1], "easier"), easier[0])

    progress_fraction = _progress_fraction(record)
    if bool(record.get("failed")) and progress_fraction is not None and progress_fraction < 0.35:
        if any(pattern in text for pattern in SEMANTIC_ONLY_PATTERNS) and "bug" in set(record.get("tags", ())):
            return AnnotationDecision(mode, "skip", 0, "failed early but annotation is semantic/bug focused")
        return AnnotationDecision(mode, "easier", 1, "failed early with low progress")

    if bool(record.get("complete")) and _health_fraction(record) >= 0.8 and "very nice" not in text:
        return AnnotationDecision(mode, "harder", 1, "completed with high remaining integrity")

    return AnnotationDecision(mode, "skip", 0, "no deterministic parameter signal")


def _build_mode_tuning(
    health: int,
    max_health: int,
    data: Mapping[str, object],
    direction: Direction,
    severity: int,
) -> ModeTuning:
    severity = max(1, min(2, int(severity)))
    health_delta = severity if direction == "easier" else -1
    next_max_health = max(1, min(100, int(max_health) + health_delta))
    next_health = max(1, min(next_max_health, int(health) + health_delta))

    next_data: dict[str, int | float] = {}
    for key, value in data.items():
        if key not in ALLOWED_DATA_KEYS or isinstance(value, bool) or not isinstance(value, (int, float)):
            continue
        if key == "target_boss_health":
            scale = 1.25 + 0.15 * (severity - 1) if direction == "easier" else 0.85
        elif key == "max_phase":
            next_data[key] = max(1, int(value) - 1) if direction == "easier" else int(value) + 1
            continue
        else:
            scale = (0.85 - 0.05 * (severity - 1)) if direction == "easier" else (1.2 + 0.1 * (severity - 1))
        low, high = DATA_RANGES[key]
        next_data[key] = _scale_value(value, scale, low, high)

    return ModeTuning(health=next_health, max_health=next_max_health, data=next_data)


def _load_latest_annotations(path: Path) -> dict[str, Mapping[str, Any]]:
    latest: dict[str, Mapping[str, Any]] = {}
    for line in path.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        record = json.loads(line)
        mode = record.get("mode")
        if isinstance(mode, str):
            latest[mode] = record
    return latest


def _normalized_feedback(record: Mapping[str, Any]) -> str:
    feedback = str(record.get("feedback", "")).lower()
    return re.sub(r"\s+", " ", feedback).strip()


def _best_match(text: str, patterns: tuple[tuple[str, int], ...]) -> tuple[str, int]:
    for pattern, severity in patterns:
        if pattern in text:
            return pattern, severity
    return "", 0


def _contains_explicit_direction(text: str) -> bool:
    return _best_match(text, EASIER_PATTERNS)[1] > 0 or _best_match(text, HARDER_PATTERNS)[1] > 0


def _effective_severity(record: Mapping[str, Any], severity: int, direction: Direction) -> int:
    if (
        direction == "easier"
        and int(record.get("health", 1)) <= 0
        and _progress_fraction(record) == 0
        and int(record.get("score", 0)) <= 0
    ):
        return max(severity, 2)
    return max(1, min(2, severity))


def _progress_fraction(record: Mapping[str, Any]) -> float | None:
    progress = record.get("progress")
    if not isinstance(progress, Mapping):
        return None
    value = progress.get("fraction")
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    return float(value)


def _health_fraction(record: Mapping[str, Any]) -> float:
    health = record.get("health")
    max_health = record.get("max_health")
    if isinstance(health, bool) or isinstance(max_health, bool) or not isinstance(health, (int, float)) or not isinstance(max_health, (int, float)):
        return 0.0
    return float(health) / max(1.0, float(max_health))


def _scale_value(value: int | float, scale: float, low: float, high: float) -> int | float:
    scaled = max(low, min(high, float(value) * scale))
    if isinstance(value, int):
        return max(1, int(round(scaled)))
    return round(scaled, 3)


if __name__ == "__main__":
    main()