"""Obiekt DATABASE silnika PIK — pliki .DTA. Wiersze rozdzielone CRLF, kolumny pionową kreską, brak nagłówka. Nazwy i typy kolumn żyją w definicji STRUCT w pliku .cnv — patrz ricsave.schema. Puste pole tekstowe zapisywane jest jako literalne NULL. """ from __future__ import annotations from pathlib import Path ENCODING = "cp1250" ROW_SEP = "\r\n" COL_SEP = "|" NULL = "NULL" class DtaError(ValueError): """Plik .DTA nie trzyma się formatu.""" class Table: """Tabela .DTA: lista wierszy, każdy to lista surowych stringów. Świadomie trzymamy surowe stringi — konwersję na int robi dopiero warstwa schematu, dzięki czemu zapis jest bajt-w-bajt identyczny gdy nic nie zmieniasz. """ def __init__(self, rows: list[list[str]], *, trailing_newline: bool = True) -> None: self.rows = rows # Pliki *_def.dta nie mają końcowego CRLF, te zapisane przez grę mają. self.trailing_newline = trailing_newline @property def width(self) -> int: return len(self.rows[0]) if self.rows else 0 @property def widths(self) -> set[int]: return {len(row) for row in self.rows} def check_width(self, expected: int, label: str = "tabela") -> None: """Waliduje szerokość wobec znanego schematu STRUCT.""" found = self.widths if found - {expected}: raise DtaError( f"{label}: oczekiwano {expected} kolumn, znaleziono {sorted(found)}" ) def __len__(self) -> int: return len(self.rows) def __getitem__(self, index: int) -> list[str]: return self.rows[index] def __repr__(self) -> str: return f"" def loads(data: bytes) -> Table: text = data.decode(ENCODING) trailing = text.endswith(ROW_SEP) body = text[: -len(ROW_SEP)] if trailing else text if not body: return Table([], trailing_newline=trailing) if "\r\n" not in body and "\n" in body: raise DtaError("wiersze rozdzielone samym LF — to nie jest plik .DTA z gry") # Bez walidacji szerokości: część statycznych tabel gry (np. fighters.dta) ma # wiersze różnej długości. Szerokość sprawdzamy tam, gdzie znamy STRUCT — # przez Table.check_width(). rows = [line.split(COL_SEP) for line in body.split(ROW_SEP)] return Table(rows, trailing_newline=trailing) def dumps(table: Table) -> bytes: text = ROW_SEP.join(COL_SEP.join(row) for row in table.rows) if table.trailing_newline and table.rows: text += ROW_SEP return text.encode(ENCODING) def read(path: str | Path) -> Table: return loads(Path(path).read_bytes()) def write(path: str | Path, table: Table) -> None: Path(path).write_bytes(dumps(table))