""" rxdata_tools.py — Operaciones sobre archivos .rxdata y PBS de PE v21.1. Filosofía: - Antigravity lee el filesystem nativamente → no reimplementamos lectura de archivos de texto (PBS, plugins, scripts .rb). - Este módulo solo expone lo que un IDE NO puede hacer por sí solo: 1. Deserializar / serializar Marshal (.rxdata) 2. Escribir PBS con validación de formato estricta 3. Templates de eventos con estructura RGSS correcta garantizada 4. Backup automático antes de cualquier escritura """ from __future__ import annotations from pathlib import Path import sys sys.path.insert(0, str(Path(__file__).parent.parent / "bridge")) from rxdata_bridge import RxDataBridge, RxDataError from event_validator import validate_event, EventValidationError # ============================================================================= # CONSTANTES RGSS # ============================================================================= CMD_SHOW_TEXT = 101 CMD_SHOW_TEXT_CONT = 401 CMD_SELF_SWITCH = 123 CMD_TRANSFER = 201 CMD_SCRIPT = 355 CMD_END = 0 # ============================================================================= # HELPERS DE COMANDOS # ============================================================================= def _cmd(code: int, indent: int = 0, params: list = None) -> dict: return {"code": code, "indent": indent, "parameters": params or []} def _end() -> dict: return _cmd(CMD_END) def _self_switch(letter: str, value: bool) -> dict: return _cmd(CMD_SELF_SWITCH, 0, [letter, 0 if value else 1]) def _text(lines: list[str]) -> list[dict]: if not lines: return [] cmds = [_cmd(CMD_SHOW_TEXT, 0, ["", 0, 2, 0])] for line in lines: cmds.append(_cmd(CMD_SHOW_TEXT_CONT, 0, [str(line)])) return cmds def _script(code: str, indent: int = 0) -> dict: return _cmd(CMD_SCRIPT, indent, [code]) def _transfer(map_id: int, x: int, y: int, direction: int = 2) -> dict: return _cmd(CMD_TRANSFER, 0, [0, map_id, x, y, direction, 0]) # ============================================================================= # HELPERS DE PÁGINAS # ============================================================================= def _condition_empty() -> dict: return { "switch1_valid": False, "switch2_valid": False, "variable_valid": False, "self_switch_valid": False, "switch1_id": 1, "switch2_id": 1, "variable_id": 1, "variable_value": 0, "self_switch_ch": "A", } def _condition_self_switch(letter: str = "A") -> dict: c = _condition_empty() c["self_switch_valid"] = True c["self_switch_ch"] = letter return c def _graphic(name: str = "", direction: int = 2) -> dict: return { "tile_id": 0, "character_name": name, "character_hue": 0, "direction": direction, "pattern": 0, "opacity": 255, "blend_type": 0, } def _move_route_default() -> dict: return { "repeat": False, "skippable": False, "list": [{"code": 0, "parameters": []}], } def _page( commands: list[dict], graphic: str = "", direction: int = 2, trigger: int = 0, move_type: int = 0, move_speed: int = 3, direction_fix: bool = False, through: bool = False, always_on_top: bool = False, condition: dict = None, ) -> dict: cmds = list(commands) if not cmds or cmds[-1]["code"] != CMD_END: cmds.append(_end()) return { "condition": condition or _condition_empty(), "graphic": _graphic(graphic, direction), "move_type": move_type, "move_speed": move_speed, "move_frequency": 3, "move_route": _move_route_default(), "walk_anime": True, "step_anime": False, "direction_fix": direction_fix, "through": through, "always_on_top": always_on_top, "trigger": trigger, "list": cmds, } def _event(name: str, x: int, y: int, pages: list[dict]) -> dict: return {"id": 0, "name": name, "x": x, "y": y, "pages": pages} # ============================================================================= # TEMPLATES DE EVENTOS # ============================================================================= def tpl_npc(name: str, x: int, y: int, graphic: str, dialogue: list[str], second_dialogue: list[str] = None, direction: int = 2) -> dict: """NPC con diálogo. second_dialogue activa self-switch A para segunda página.""" cmds1 = _text(dialogue) if second_dialogue: cmds1.append(_self_switch("A", True)) pages = [_page(cmds1, graphic=graphic, direction=direction)] if second_dialogue: pages.append(_page( _text(second_dialogue), graphic=graphic, direction=direction, condition=_condition_self_switch("A"), )) return _event(name, x, y, pages) def tpl_item_ball(name: str, x: int, y: int, item: str) -> dict: """Objeto recogible en el suelo. Desaparece tras recogerlo.""" p1 = _page([_script(f"pbItemBall(:{item})"), _self_switch("A", True)], trigger=1) p2 = _page([_end()], trigger=1, through=True, condition=_condition_self_switch("A")) return _event(name, x, y, [p1, p2]) def tpl_mart(name: str, x: int, y: int, graphic: str, items: list[str], intro: list[str] = None) -> dict: """Tienda Pokémon.""" items_str = ", ".join(f":{i}" for i in items) cmds = _text(intro or ["Welcome! Take your time."]) cmds.append(_script(f"pbPokemonMart([{items_str}])")) return _event(name, x, y, [_page(cmds, graphic=graphic)]) def tpl_warp(name: str, x: int, y: int, dest_map_id: int, dest_x: int, dest_y: int, direction: int = 2) -> dict: """Warp/puerta que transfiere al jugador.""" return _event(name, x, y, [ _page([_transfer(dest_map_id, dest_x, dest_y, direction)], trigger=1) ]) def tpl_script(name: str, x: int, y: int, ruby_code: str, trigger: int = 0, graphic: str = "", one_time: bool = False) -> dict: """Evento que ejecuta código Ruby de PE v21.1.""" cmds = [_script(ruby_code)] if one_time: cmds.append(_self_switch("A", True)) pages = [_page(cmds, graphic=graphic, trigger=trigger)] if one_time: pages.append(_page([_end()], trigger=trigger, through=True, condition=_condition_self_switch("A"))) return _event(name, x, y, pages) def tpl_multipage_script(name: str, x: int, y: int, pages_config: list[dict]) -> dict: """ Evento con múltiples páginas controladas por el usuario. Cada página en pages_config puede tener: - ruby_code (str): el script a ejecutar - graphic (str): gráfico de la página - trigger (int): detonante (0=Action, 1=Touch, etc) - condition (dict): {'self_switch': 'A'} opcional - through (bool) - direction (int) """ pages = [] for pcfg in pages_config: cmds = [] if "ruby_code" in pcfg and pcfg["ruby_code"]: cmds.append(_script(pcfg["ruby_code"])) condition = _condition_empty() if "condition" in pcfg and "self_switch" in pcfg["condition"]: condition = _condition_self_switch(pcfg["condition"]["self_switch"]) pages.append(_page( cmds, graphic=pcfg.get("graphic", ""), direction=pcfg.get("direction", 2), trigger=pcfg.get("trigger", 0), through=pcfg.get("through", False), condition=condition )) return _event(name, x, y, pages) # ============================================================================= # PBS WRITER — solo escritura con validación # Antigravity lee PBS/pokemon.txt directamente; nosotros solo validamos y escribimos. # ============================================================================= _VALID_TYPES = { "NORMAL","FIRE","WATER","ELECTRIC","GRASS","ICE","FIGHTING","POISON", "GROUND","FLYING","PSYCHIC","BUG","ROCK","GHOST","DRAGON","DARK","STEEL","FAIRY", } class PBSWriter: """ Escribe entradas en archivos PBS con validación estricta de formato. NO reimplementa lectura — Antigravity lee los .txt directamente. """ def __init__(self, project_path: str | Path): self.pbs = Path(project_path) / "PBS" if not self.pbs.exists(): raise ValueError(f"Carpeta PBS no encontrada en {project_path}") def _read_raw(self, filename: str) -> tuple[list[str], dict[str, int]]: """Lee el archivo y devuelve (líneas, índice de bloques).""" path = self.pbs / filename lines = path.read_text(encoding="utf-8", errors="replace").splitlines() index = {} for i, line in enumerate(lines): s = line.strip() if s.startswith("[") and s.endswith("]"): index[s[1:-1].upper()] = i return lines, index def _write_raw(self, filename: str, lines: list[str]): path = self.pbs / filename path.write_text("\n".join(lines) + "\n", encoding="utf-8") def _replace_or_append_block(self, filename: str, key: str, new_fields: dict): """Actualiza un bloque existente o lo añade al final del archivo. Preserva los campos no modificados.""" lines, index = self._read_raw(filename) key_up = key.upper() if key_up in index: start = index[key_up] # Encontrar fin del bloque (siguiente [KEY] o EOF) end = len(lines) for i in range(start + 1, len(lines)): s = lines[i].strip() if s.startswith("[") and s.endswith("]"): end = i break # Extraer las líneas y buscar los campos existentes block_lines = lines[start:end] existing_fields = {} for i, line in enumerate(block_lines): if "=" in line: k, v = line.split("=", 1) existing_fields[k.strip()] = i # Actualizar campos existentes o añadir al bloque for k, v in new_fields.items(): if k in existing_fields: line_idx = existing_fields[k] block_lines[line_idx] = f"{k} = {v}" else: block_lines.append(f"{k} = {v}") lines[start:end] = block_lines else: new_block = [f"[{key_up}]"] for k, v in new_fields.items(): new_block.append(f"{k} = {v}") if lines and lines[-1].strip(): lines.append("") lines.extend(new_block) self._write_raw(filename, lines) # ── Pokémon ─────────────────────────────────────────────────────────────── def write_pokemon(self, species: str, fields: dict) -> str: """Escribe o actualiza una especie en pokemon.txt con validación.""" self._validate_pokemon(fields) self._replace_or_append_block("pokemon.txt", species, fields) return f"Especie {species.upper()} escrita en PBS/pokemon.txt" def _validate_pokemon(self, fields: dict): if "Type1" in fields and fields["Type1"].upper() not in _VALID_TYPES: raise ValueError(f"Type1 inválido: '{fields['Type1']}'. Válidos: {sorted(_VALID_TYPES)}") if "Type2" in fields and fields["Type2"].upper() not in _VALID_TYPES: raise ValueError(f"Type2 inválido: '{fields['Type2']}'") if "BaseStats" in fields: parts = str(fields["BaseStats"]).split(",") if len(parts) != 6 or not all(p.strip().isdigit() for p in parts): raise ValueError(f"BaseStats debe ser exactamente 6 números: '{fields['BaseStats']}'") if "Height" in fields: try: float(fields["Height"]) except ValueError: raise ValueError(f"Height debe ser un número decimal: '{fields['Height']}'") if "Weight" in fields: try: float(fields["Weight"]) except ValueError: raise ValueError(f"Weight debe ser un número decimal: '{fields['Weight']}'") # ── Moves ───────────────────────────────────────────────────────────────── def write_move(self, move: str, fields: dict) -> str: """Escribe o actualiza un movimiento en moves.txt.""" if "Type" in fields and fields["Type"].upper() not in _VALID_TYPES: raise ValueError(f"Type inválido: '{fields['Type']}'") if "Category" in fields and fields["Category"] not in ("Physical", "Special", "Status"): raise ValueError(f"Category debe ser Physical, Special o Status") self._replace_or_append_block("moves.txt", move, fields) return f"Movimiento {move.upper()} escrito en PBS/moves.txt" # ── Items ───────────────────────────────────────────────────────────────── def write_item(self, item: str, fields: dict) -> str: """Escribe o actualiza un objeto en items.txt.""" if "Price" in fields: try: int(fields["Price"]) except ValueError: raise ValueError(f"Price debe ser un entero: '{fields['Price']}'") self._replace_or_append_block("items.txt", item, fields) return f"Objeto {item.upper()} escrito en PBS/items.txt" # ============================================================================= # RXDATA TOOLS — clase principal # ============================================================================= class RxDataTools: """ Única interfaz MCP para operaciones sobre .rxdata. Solo expone lo que Antigravity no puede hacer nativamente. """ def __init__(self, project_path: str | Path, ruby_path: str = "ruby"): self.bridge = RxDataBridge(project_path, ruby_path) self.pbs = PBSWriter(project_path) # ── Mapas — lectura ─────────────────────────────────────────────────────── def list_maps(self) -> list[dict]: return self.bridge.list_maps() def get_map_info(self, map_id: int) -> dict: m = self.bridge.read_map(map_id) events = m.get("events") or {} return { "map_id": map_id, "width": m.get("width"), "height": m.get("height"), "tileset": m.get("tileset_id"), "n_events": len(events), "events": [ {"id": eid, "name": ev.get("name",""), "x": ev.get("x"), "y": ev.get("y")} for eid, ev in events.items() if isinstance(ev, dict) ], } def get_map_section(self, map_id: int, x1: int, y1: int, x2: int, y2: int) -> dict: return self.bridge.get_map_section(map_id, x1, y1, x2, y2) def find_free_positions(self, map_id: int, count: int = 10) -> list[dict]: """Posiciones del mapa sin eventos.""" m = self.bridge.read_map(map_id) events = m.get("events") or {} occupied = {(ev.get("x"), ev.get("y")) for ev in events.values() if isinstance(ev, dict)} free = [] for y in range(1, m.get("height", 15) - 1): for x in range(1, m.get("width", 20) - 1): if (x, y) not in occupied: free.append({"x": x, "y": y}) if len(free) >= count: return free return free # ── Mapas — escritura ───────────────────────────────────────────────────── def create_map(self, name: str, width: int, height: int, tileset_id: int = 1) -> int: return self.bridge.create_map(name, width, height, tileset_id) # ── Eventos — templates ─────────────────────────────────────────────────── def _add_validated(self, map_id: int, event: dict) -> int: """Único punto de escritura de eventos. Valida antes de tocar el disco.""" try: validate_event(event, map_id) except EventValidationError as e: raise RxDataError(f"Evento inválido — no se escribirá nada:\n{e}") return self.bridge.add_event_to_map(map_id, event) def add_npc(self, map_id: int, name: str, x: int, y: int, graphic: str, dialogue: list[str], second_dialogue: list[str] = None, direction: int = 2) -> int: ev = tpl_npc(name, x, y, graphic, dialogue, second_dialogue, direction) return self._add_validated(map_id, ev) def add_item_ball(self, map_id: int, name: str, x: int, y: int, item: str) -> int: return self._add_validated(map_id, tpl_item_ball(name, x, y, item)) def add_mart(self, map_id: int, name: str, x: int, y: int, graphic: str, items: list[str], intro: list[str] = None) -> int: return self._add_validated(map_id, tpl_mart(name, x, y, graphic, items, intro)) def add_warp(self, map_id: int, name: str, x: int, y: int, dest_map_id: int, dest_x: int, dest_y: int, direction: int = 2) -> int: ev = tpl_warp(name, x, y, dest_map_id, dest_x, dest_y, direction) return self._add_validated(map_id, ev) def add_script_event(self, map_id: int, name: str, x: int, y: int, ruby_code: str, trigger: int = 0, graphic: str = "", one_time: bool = False) -> int: ev = tpl_script(name, x, y, ruby_code, trigger, graphic, one_time) return self._add_validated(map_id, ev) def add_multipage_event(self, map_id: int, name: str, x: int, y: int, pages_config: list[dict]) -> int: ev = tpl_multipage_script(name, x, y, pages_config) return self._add_validated(map_id, ev) def add_raw_event(self, map_id: int, event_data: dict) -> int: """Añade un evento desde dict completo — siempre pasa por validación.""" return self._add_validated(map_id, event_data) # ── Eventos — edición ───────────────────────────────────────────────────── def update_dialogue(self, map_id: int, event_id: int, new_dialogue: list[str], page: int = 0) -> str: m = self.bridge.read_map(map_id) events = m.get("events") or {} key = str(event_id) if key not in events: raise RxDataError(f"Evento {event_id} no existe en mapa {map_id}") pages = events[key].get("pages") or [] if page >= len(pages): raise RxDataError(f"Página {page} no existe en evento {event_id}") old = pages[page].get("list") or [] other = [c for c in old if isinstance(c, dict) and c.get("code") not in (CMD_SHOW_TEXT, CMD_SHOW_TEXT_CONT, CMD_END)] pages[page]["list"] = _text(new_dialogue) + other + [_end()] self.bridge.write_map(map_id, m) return f"Diálogo del evento {event_id} (página {page}) actualizado." def edit_event_page(self, map_id: int, event_id: int, page: int, changes: dict) -> str: self.bridge.edit_event(map_id, event_id, page, changes) return f"Evento {event_id} página {page} actualizado en mapa {map_id}." def delete_event(self, map_id: int, event_id: int) -> str: self.bridge.delete_event(map_id, event_id) return f"Evento {event_id} eliminado del mapa {map_id}." # ── PBS — escritura con validación ──────────────────────────────────────── def write_pokemon(self, species: str, fields: dict) -> str: return self.pbs.write_pokemon(species, fields) def write_move(self, move: str, fields: dict) -> str: return self.pbs.write_move(move, fields) def write_item(self, item: str, fields: dict) -> str: return self.pbs.write_item(item, fields) # ── Utilidades ──────────────────────────────────────────────────────────── def undo(self) -> str: return self.bridge.undo_latest()