pgames-manager/backend/app/converter/atl.py

"""Ren'Py ATL (animation and transformation language) → PGVN timeline.

A block becomes ``{"steps": [...], "repeat": None | True | n}``. Steps:

* ``{"op": "set", "props": {...}}`` – assign properties at once
* ``{"op": "move", "warp": "linear|ease|easein|easeout", "t": seconds, "props": {...}}``
* ``{"op": "pause", "t": seconds}``
* ``{"op": "parallel", "blocks": [block, ...]}``
* ``{"op": "choice", "blocks": [{"w": weight, "block": block}, ...]}``
* ``{"op": "block", "block": block}``
* ``{"op": "on", "events": [...], "block": block}``
* ``{"op": "use", "name": transform}``

A number or property value that is not a literal becomes ``{"e": expr}`` and is evaluated when the
timeline runs, with the transform's parameters in scope (``transform blink(delay, cycle):``).

Everything else (``function``, ``contains``, ``time``, expressions, shaders, matrix
colours) is reported through ``warn`` so the game is not imported half-working.
"""
import ast
import re
from collections.abc import Callable
from typing import Any

from app.converter import pyexpr
from app.converter.rpy_helpers import POSITION_PROPS
from app.converter.rpy_lexer import Line

Block = dict[str, Any]

# Warpers differ only in easing shape; the player implements these four curves and
# maps the named variants (easein_quad, ease_back, ...) onto the closest one.
WARPER = re.compile(r"^(linear|ease|easein|easeout|easeinout)(_(quad|cubic|quart|quint|expo|circ|back|bounce|elastic|sine))?$")
IGNORED = {"subpixel", "animation"}  # no visible effect in the browser
EVENTS = {"show", "hide", "replace", "replaced", "start", "update", "appear", "default"}


def tokens(text: str) -> list[str]:
    """Split on whitespace/commas outside brackets and quotes."""
    out, cur, depth, quote = [], [], 0, ""
    for ch in text:
        if quote:
            cur.append(ch)
            if ch == quote:
                quote = ""
        elif ch in "\"'":
            quote = ch
            cur.append(ch)
        elif ch in "([{":
            depth += 1
            cur.append(ch)
        elif ch in ")]}":
            depth -= 1
            cur.append(ch)
        elif (ch.isspace() or ch == ",") and depth == 0:
            if cur:
                out.append("".join(cur))
                cur = []
        else:
            cur.append(ch)
    if cur:
        out.append("".join(cur))
    return out


class AtlParser:
    def __init__(self, warn: Callable[[str, Line], None], is_static: Callable[[str], dict | None],
                 on_use: Callable[[str], None] = lambda name: None):
        self.warn = warn
        self.on_use = on_use
        # name -> props when `name` is a known static transform (inlined), else None
        self.static_transform = is_static

    # --- helpers -------------------------------------------------------------------

    def value(self, raw: str, line: Line) -> Any:
        """A literal, or ``{"e": expr}`` for anything computed at run time."""
        try:
            return ast.literal_eval(raw)
        except (ValueError, SyntaxError):
            pass
        try:
            return {"e": pyexpr.parse_expr(raw)}
        except pyexpr.Unsupported:
            self.warn(f"Не поддерживается ATL: выражение {raw}", line)
            return None

    def number(self, raw: str, line: Line) -> Any:
        value = self.value(raw, line)
        if value is None:
            return None
        if isinstance(value, dict):
            return value
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            self.warn(f"Не поддерживается ATL: значение {raw}", line)
            return None
        return float(value)

    def props(self, parts: list[str], line: Line) -> dict[str, Any]:
        out: dict[str, Any] = {}
        i = 0
        while i < len(parts):
            key = parts[i]
            if key in IGNORED:
                i += 2
                continue
            if key not in POSITION_PROPS:
                self.warn(f"Не поддерживается ATL: свойство {key}", line)
                return out
            if i + 1 >= len(parts):
                self.warn(f"Не поддерживается ATL: у {key} нет значения", line)
                return out
            value = self.value(parts[i + 1], line)
            if value is None:
                return out
            out[key] = list(value) if isinstance(value, tuple) else value
            i += 2
        return out

    # --- block ----------------------------------------------------------------------

    def block(self, lines: list[Line]) -> Block:
        steps: list[dict[str, Any]] = []
        repeat: Any = None
        i = 0
        while i < len(lines):
            line = lines[i]
            parts = tokens(line.text.rstrip(":").strip() if line.block else line.text.strip())
            if not parts:
                i += 1
                continue
            first = parts[0]
            i += 1
            if first == "repeat":
                repeat = True if len(parts) == 1 else self.number(parts[1], line)
                if repeat is not True and repeat is not None:
                    repeat = repeat if isinstance(repeat, dict) else int(repeat)
            elif first == "pause":
                seconds = self.number(parts[1], line) if len(parts) > 1 else None
                if seconds is not None:
                    steps.append({"op": "pause", "t": seconds})
            elif first in ("parallel", "choice", "on", "block", "time", "contains", "function"):
                i = self.compound(first, parts, lines, i - 1, steps)
            elif WARPER.match(first):
                self.move(first, parts, line, steps)
            elif first in IGNORED:
                continue
            elif first in POSITION_PROPS:
                props = self.props(parts, line)
                if props:
                    steps.append({"op": "set", "props": props})
            elif len(parts) == 1 and re.fullmatch(r"[A-Za-z_]\w*", first):
                self.on_use(first)
                known = self.static_transform(first)
                steps.append({"op": "set", "props": known} if known is not None else {"op": "use", "name": first})
            else:
                self.warn(f"Не поддерживается ATL: {first}", line)
        return {"steps": steps, "repeat": repeat}

    def move(self, warp: str, parts: list[str], line: Line, steps: list) -> None:
        base = WARPER.match(warp).group(1)  # type: ignore[union-attr]
        if len(parts) < 2:
            self.warn(f"Не поддерживается ATL: у {warp} нет длительности", line)
            return
        seconds = self.number(parts[1], line)
        props = self.props(parts[2:], line)
        for child in line.block:  # `linear 1.0:` followed by indented property lines
            props.update(self.props(tokens(child.text), child))
        if seconds is None or not props:
            return
        steps.append({"op": "move", "warp": "ease" if base == "easeinout" else base, "t": seconds, "props": props})

    def compound(self, first: str, parts: list[str], lines: list[Line], index: int, steps: list) -> int:
        """Handle parallel/choice/on groups. Returns the index of the next unread line."""
        line = lines[index]
        if first in ("time", "contains", "function"):
            self.warn(f"Не поддерживается ATL: {first}", line)
            return index + 1
        if first == "block":
            steps.append({"op": "block", "block": self.block(line.block)})
            return index + 1
        if first == "on":
            events = [e for e in parts[1:] if e]
            if not events or any(e not in EVENTS for e in events):
                self.warn(f"Не поддерживается ATL: событие {' '.join(events)}", line)
            steps.append({"op": "on", "events": events, "block": self.block(line.block)})
            return index + 1
        # consecutive `parallel:` / `choice:` clauses form one statement
        group = []
        j = index
        while j < len(lines):
            current = lines[j]
            words = tokens(current.text.rstrip(":").strip())
            if not words or words[0] != first:
                break
            weight: Any = 1.0
            if first == "choice" and len(words) > 1:
                weight = self.number(words[1], current) or 1.0
            group.append((weight, self.block(current.block)))
            j += 1
        if first == "parallel":
            steps.append({"op": "parallel", "blocks": [b for _, b in group]})
        else:
            steps.append({"op": "choice", "blocks": [{"w": w, "block": b} for w, b in group]})
        return j


def computed(value: Any) -> bool:
    if isinstance(value, dict):
        return "e" in value
    return isinstance(value, list) and any(computed(item) for item in value)


def static_props(block: Block) -> dict[str, Any] | None:
    """Merged properties when the block only assigns, else None (it needs the timeline)."""
    if block["repeat"] is not None or any(step["op"] != "set" for step in block["steps"]):
        return None
    if any(computed(v) for step in block["steps"] for v in step["props"].values()):
        return None
    merged: dict[str, Any] = {}
    for step in block["steps"]:
        merged.update(step["props"])
    return merged


def initial_props(block: Block) -> dict[str, Any]:
    """Properties assigned before the first animated step (the state the element is created in)."""
    merged: dict[str, Any] = {}
    for step in block["steps"]:
        if step["op"] != "set":
            break
        merged.update({k: v for k, v in step["props"].items() if not computed(v)})
    return merged