Merge gfx/optional (asset-free sky & grading builder) into integration
This commit is contained in:
pkuzic
@@ -43,6 +43,8 @@ the reference 1.12 slots, so their fidelity fixes remain active in Classic: auth
|
||||
alpha tracks affect the layer, non-white M2 colours reach the mesh, deterministic captures pose at
|
||||
their pinned time, and the batch-order base reserves the lower celestial band and final fog cone.
|
||||
|
||||
Asset-free local builder and bilingual install guide: [`Optional/sky-and-grading/`](Optional/sky-and-grading/README.md).
|
||||
|
||||
Players reach all of it from **Options -> Advanced Graphics** (a Graphics Preset over everything, a Render Distance slider, a Lighting Quality preset Off / Low / Medium / High / Ultra plus the individual rows; Classic / Off is the original client look). The dev build has a panel for all of it: **Ctrl+Shift+D → Lighting & shadows**, with Dim / Default /
|
||||
Bright presets.
|
||||
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
# Optional skyboxes and colour grading
|
||||
|
||||
This asset-free kit builds an optional MPQ on your machine. It adds zone
|
||||
skyboxes for Burning Steppes, Blasted Lands, and Mount Hyjal; project-authored
|
||||
per-zone colour grading for Elwynn Forest, Duskwood, Westfall, and Redridge;
|
||||
and, with `--wf-colours`, a moderated import of night colours from your own WoW
|
||||
Forever installation.
|
||||
|
||||
The repository contains only scripts and benilla's own grading LUTs/table. The
|
||||
Blizzard models, textures, and lighting tables are read from installations you
|
||||
own and appear only in your local build directory.
|
||||
|
||||
## Requirements
|
||||
|
||||
- Windows and Python 3.8 or newer.
|
||||
- A World of Warcraft 1.12.1 client and its `Data` directory.
|
||||
- For skyboxes, a local WoW Forever `_classic_beta_` or current retail install.
|
||||
Point the command at the shared World of Warcraft root (the directory with
|
||||
`.build.info` and `Data`), or at `_classic_beta_` itself.
|
||||
- Git and the external `wowdev/pywowlib` checkout pinned below. It is not
|
||||
redistributed here.
|
||||
- 64-bit StormLib 9.40 (`StormLib.dll`) from the official
|
||||
[StormLib releases](https://github.com/ladislav-zezula/StormLib/releases/tag/v9.40).
|
||||
Download `stormlib_dll.zip` and use its x64 DLL. StormLib is not redistributed.
|
||||
- Internet access on the first run for the public wow-listfile and public TACT
|
||||
key list, unless `--listfile` and `--tact-keys` name local copies.
|
||||
|
||||
## Install, one command per step
|
||||
|
||||
Run these in PowerShell. Change the example paths for your machine.
|
||||
|
||||
1. Clone the exact pywowlib revision used by the converter.
|
||||
|
||||
```powershell
|
||||
git clone https://github.com/wowdev/pywowlib.git C:\Tools\pywowlib
|
||||
```
|
||||
|
||||
2. Pin it to the tested commit.
|
||||
|
||||
```powershell
|
||||
git -C C:\Tools\pywowlib checkout 55276dc5c2195da7fe136638a2a59716622f8c65
|
||||
```
|
||||
|
||||
3. Install the Python modules needed by the M2 reader. pywowlib remains an
|
||||
external checkout; this installs dependencies only.
|
||||
|
||||
```powershell
|
||||
py -3 -m pip install bidict==0.23.1 multimethod numpy
|
||||
```
|
||||
|
||||
4. Download StormLib 9.40's `stormlib_dll.zip` from the release linked above,
|
||||
extract the 64-bit `StormLib.dll`, and place it at
|
||||
`C:\Tools\StormLib\StormLib.dll`.
|
||||
|
||||
5. Close the game and build the candidate. The tool never writes to either
|
||||
input installation.
|
||||
|
||||
```powershell
|
||||
py -3 .\tools\build_sky_patch.py --wow-forever "C:\Games\World of Warcraft" --client-data "C:\Games\WoW-1.12.1\Data" --pywowlib "C:\Tools\pywowlib" --stormlib "C:\Tools\StormLib\StormLib.dll" --out "C:\Temp\benilla-sky"
|
||||
```
|
||||
|
||||
6. Verify an isolated Data copy before installation. `C:\Temp\WoW-test\Data`
|
||||
should be a copy or hard-link mirror in which only `patch-Z.mpq` is an
|
||||
ordinary copy; replace that copy with the newly built archive first.
|
||||
|
||||
```powershell
|
||||
py -3 .\tools\verify_client_chain.py "C:\Temp\WoW-test\Data" --stage "C:\Temp\benilla-sky\stage" --stormlib "C:\Tools\StormLib\StormLib.dll"
|
||||
```
|
||||
|
||||
7. Back up the live target archive if one exists.
|
||||
|
||||
```powershell
|
||||
Copy-Item "C:\Games\WoW-1.12.1\Data\patch-Z.mpq" "C:\Games\WoW-1.12.1\Data\patch-Z.before-benilla.mpq.bak"
|
||||
```
|
||||
|
||||
8. Copy the verified candidate into the client.
|
||||
|
||||
```powershell
|
||||
Copy-Item -Force "C:\Temp\benilla-sky\patch-Z.MPQ" "C:\Games\WoW-1.12.1\Data\patch-Z.mpq"
|
||||
```
|
||||
|
||||
If no `patch-Z.mpq` existed, skip step 7. The build report's `archive_base`
|
||||
field records whether the builder preserved an existing patch-Z.
|
||||
|
||||
### Variants
|
||||
|
||||
Build only the project-owned grading table and LUTs; this needs no modern WoW
|
||||
install or pywowlib:
|
||||
|
||||
```powershell
|
||||
py -3 .\tools\build_sky_patch.py --grading-only --client-data "C:\Games\WoW-1.12.1\Data" --stormlib "C:\Tools\StormLib\StormLib.dll" --out "C:\Temp\benilla-grading"
|
||||
```
|
||||
|
||||
Add moderated WoW Forever night colours. LightData is extracted directly from
|
||||
your installation; no derived LightData is shipped here:
|
||||
|
||||
```powershell
|
||||
py -3 .\tools\build_sky_patch.py --wow-forever "C:\Games\World of Warcraft" --client-data "C:\Games\WoW-1.12.1\Data" --pywowlib "C:\Tools\pywowlib" --stormlib "C:\Tools\StormLib\StormLib.dll" --wf-colours --out "C:\Temp\benilla-sky-wf"
|
||||
```
|
||||
|
||||
## Enable in game
|
||||
|
||||
Open **Options → Advanced Graphics** and enable **Zone Skyboxes** and
|
||||
**Colour Grading**. The equivalent console commands are:
|
||||
|
||||
```text
|
||||
/console zoneSkyboxes 1
|
||||
/console colorGrading 1
|
||||
```
|
||||
|
||||
Use `0` instead of `1` to disable either effect without removing the MPQ.
|
||||
|
||||
## MPQ load order: read before copying
|
||||
|
||||
All five mutually dependent lighting DBCs must win from the same, highest
|
||||
letter patch that carries any of them. The default is `patch-Z.mpq`. If the
|
||||
client already has patch-Z, the builder copies that archive and overlays the
|
||||
new files so unrelated members survive; install the resulting candidate as a
|
||||
replacement, not as a second archive.
|
||||
|
||||
Do not blindly rename the output. A later-mounted archive that also contains
|
||||
`Light.dbc`, `LightParams.dbc`, `LightIntBand.dbc`, `LightFloatBand.dbc`, or
|
||||
`LightSkybox.dbc` can split the table set and break the sky. Some custom clients
|
||||
use punctuation after ordinary letters; in that case merge this stage into the
|
||||
actual last patch or remove the conflicting DBC copies, then run
|
||||
`verify_client_chain.py --require-winner ANY`. A conventional patch-Z setup
|
||||
should keep the default strict check.
|
||||
|
||||
The extended four-field `LightSkybox.dbc` is for benilla. Do not use this patch
|
||||
with an unmodified stock 1.12 executable that expects the two-field table.
|
||||
|
||||
## Uninstall
|
||||
|
||||
Close the game. If the build preserved an existing patch-Z, restore the backup:
|
||||
|
||||
```powershell
|
||||
Move-Item -Force "C:\Games\WoW-1.12.1\Data\patch-Z.before-benilla.mpq.bak" "C:\Games\WoW-1.12.1\Data\patch-Z.mpq"
|
||||
```
|
||||
|
||||
If there was no prior patch-Z and this kit created it, remove only that exact
|
||||
archive:
|
||||
|
||||
```powershell
|
||||
Remove-Item -LiteralPath "C:\Games\WoW-1.12.1\Data\patch-Z.mpq"
|
||||
```
|
||||
|
||||
Deleting a pre-existing patch-Z would remove unrelated custom content, so use
|
||||
`build-report.json` and your backup to distinguish these cases.
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
- **pywowlib revision mismatch:** run the checkout command from step 2. The
|
||||
builder deliberately rejects another revision so M2 layout behaviour cannot
|
||||
drift silently.
|
||||
- **StormLib not found:** pass `--stormlib` or set `STORMLIB_PATH` to the x64
|
||||
DLL. A 32-bit DLL cannot load into 64-bit Python.
|
||||
- **No CASC build/config:** point `--wow-forever` at the World of Warcraft root
|
||||
with `.build.info` and `Data`, or at its `_classic_beta_` child.
|
||||
- **Missing/encrypted CASC block:** update the public TACT key file or rerun
|
||||
without a stale `--tact-keys` copy. The tool uses public keys only.
|
||||
- **Unexpected source schema/clone start:** another addon has already changed
|
||||
the winning lighting DBCs. Remove that conflict or merge deliberately; the
|
||||
tool refuses to guess.
|
||||
- **Sky does not appear:** enable `zoneSkyboxes`, fully restart the client, and
|
||||
run `verify_client_chain.py` against the actual Data directory.
|
||||
- **Bands fail:** run `check_bands.py <Data directory> --stormlib <DLL>`. Correct
|
||||
clone keys are `(P-1)*18+b+1` and `(P-1)*6+b+1`.
|
||||
- **Output already exists:** choose a fresh `--out`, or pass `--force` only when
|
||||
replacing the candidate archive in that output directory is intended.
|
||||
|
||||
## Credits and rights
|
||||
|
||||
WoW Forever and retail skybox art is © Blizzard Entertainment and is extracted
|
||||
locally by the user from their own installation. No Blizzard art or client DBC
|
||||
is distributed here.
|
||||
|
||||
The colour-grading technique is based on WarcraftXL's `wxl-retail-grading` work
|
||||
by iThorgrim, used with attribution. The grading LUTs and `MonkeyZoneGrade.dbc`
|
||||
in `data/` are original benilla project data, licensed MIT OR Apache-2.0.
|
||||
@@ -0,0 +1,167 @@
|
||||
# Дополнительные небеса и цветокоррекция
|
||||
|
||||
Этот комплект без игровых ресурсов собирает MPQ локально. Он добавляет небеса
|
||||
зон для Burning Steppes, Blasted Lands и Mount Hyjal, собственную
|
||||
цветокоррекцию benilla для Elwynn Forest, Duskwood, Westfall и Redridge, а с
|
||||
ключом `--wf-colours` — умеренно смешанные ночные цвета из вашей установки WoW
|
||||
Forever.
|
||||
|
||||
В репозитории находятся только скрипты и созданные проектом LUT/таблица. Модели,
|
||||
текстуры и исходные DBC Blizzard читаются из ваших установок и записываются
|
||||
только во внешний каталог сборки.
|
||||
|
||||
## Требования
|
||||
|
||||
- Windows и Python 3.8 или новее.
|
||||
- Клиент World of Warcraft 1.12.1 и его каталог `Data`.
|
||||
- Для небес — локальная установка WoW Forever `_classic_beta_` либо retail.
|
||||
- Git и внешний `wowdev/pywowlib` точной ревизии ниже.
|
||||
- 64-битный StormLib 9.40 (`StormLib.dll`) из официальных
|
||||
[релизов StormLib](https://github.com/ladislav-zezula/StormLib/releases/tag/v9.40).
|
||||
Скачайте `stormlib_dll.zip` и используйте x64 DLL.
|
||||
- Интернет при первом запуске для публичного listfile и списка публичных TACT
|
||||
ключей, либо локальные файлы через `--listfile` и `--tact-keys`.
|
||||
|
||||
## Установка — одна команда на шаг
|
||||
|
||||
Команды выполняются в PowerShell; замените пути на свои.
|
||||
|
||||
1. Клонируйте pywowlib.
|
||||
|
||||
```powershell
|
||||
git clone https://github.com/wowdev/pywowlib.git C:\Tools\pywowlib
|
||||
```
|
||||
|
||||
2. Выберите проверенную ревизию.
|
||||
|
||||
```powershell
|
||||
git -C C:\Tools\pywowlib checkout 55276dc5c2195da7fe136638a2a59716622f8c65
|
||||
```
|
||||
|
||||
3. Установите зависимости Python. Сам pywowlib остаётся внешним checkout.
|
||||
|
||||
```powershell
|
||||
py -3 -m pip install bidict==0.23.1 multimethod numpy
|
||||
```
|
||||
|
||||
4. Скачайте `stormlib_dll.zip` версии 9.40 по ссылке выше и распакуйте x64
|
||||
`StormLib.dll` в `C:\Tools\StormLib\StormLib.dll`.
|
||||
|
||||
5. Закройте игру и соберите кандидат. Скрипт не пишет в исходные установки.
|
||||
|
||||
```powershell
|
||||
py -3 .\tools\build_sky_patch.py --wow-forever "C:\Games\World of Warcraft" --client-data "C:\Games\WoW-1.12.1\Data" --pywowlib "C:\Tools\pywowlib" --stormlib "C:\Tools\StormLib\StormLib.dll" --out "C:\Temp\benilla-sky"
|
||||
```
|
||||
|
||||
6. Сначала проверьте изолированную копию `Data`. Все файлы можно сделать
|
||||
жёсткими ссылками, но `patch-Z.mpq` должен быть обычной копией; замените его
|
||||
собранным кандидатом и выполните:
|
||||
|
||||
```powershell
|
||||
py -3 .\tools\verify_client_chain.py "C:\Temp\WoW-test\Data" --stage "C:\Temp\benilla-sky\stage" --stormlib "C:\Tools\StormLib\StormLib.dll"
|
||||
```
|
||||
|
||||
7. Если живой `patch-Z.mpq` уже есть, создайте резервную копию.
|
||||
|
||||
```powershell
|
||||
Copy-Item "C:\Games\WoW-1.12.1\Data\patch-Z.mpq" "C:\Games\WoW-1.12.1\Data\patch-Z.before-benilla.mpq.bak"
|
||||
```
|
||||
|
||||
8. Установите проверенный кандидат.
|
||||
|
||||
```powershell
|
||||
Copy-Item -Force "C:\Temp\benilla-sky\patch-Z.MPQ" "C:\Games\WoW-1.12.1\Data\patch-Z.mpq"
|
||||
```
|
||||
|
||||
Если patch-Z раньше не существовал, пропустите шаг 7. Поле `archive_base` в
|
||||
`build-report.json` показывает, был ли старый архив сохранён внутри кандидата.
|
||||
|
||||
### Варианты
|
||||
|
||||
Только собственная цветокоррекция, без современной установки и pywowlib:
|
||||
|
||||
```powershell
|
||||
py -3 .\tools\build_sky_patch.py --grading-only --client-data "C:\Games\WoW-1.12.1\Data" --stormlib "C:\Tools\StormLib\StormLib.dll" --out "C:\Temp\benilla-grading"
|
||||
```
|
||||
|
||||
Небеса и умеренные ночные цвета WoW Forever, извлечённые из вашей LightData:
|
||||
|
||||
```powershell
|
||||
py -3 .\tools\build_sky_patch.py --wow-forever "C:\Games\World of Warcraft" --client-data "C:\Games\WoW-1.12.1\Data" --pywowlib "C:\Tools\pywowlib" --stormlib "C:\Tools\StormLib\StormLib.dll" --wf-colours --out "C:\Temp\benilla-sky-wf"
|
||||
```
|
||||
|
||||
## Включение в игре
|
||||
|
||||
Включите **Zone Skyboxes** и **Colour Grading** в **Options → Advanced
|
||||
Graphics**. Те же настройки через консоль:
|
||||
|
||||
```text
|
||||
/console zoneSkyboxes 1
|
||||
/console colorGrading 1
|
||||
```
|
||||
|
||||
Значение `0` отключает эффект без удаления MPQ.
|
||||
|
||||
## Порядок загрузки MPQ
|
||||
|
||||
Все пять связанных DBC освещения должны загружаться из одного самого старшего
|
||||
буквенного патча, который содержит любой из них. По умолчанию это
|
||||
`patch-Z.mpq`. Если он уже существует, сборщик копирует его и накладывает новые
|
||||
файлы, сохраняя постороннее содержимое; результат устанавливается вместо
|
||||
старого архива.
|
||||
|
||||
Не переименовывайте результат вслепую. Более поздний архив с `Light.dbc`,
|
||||
`LightParams.dbc`, `LightIntBand.dbc`, `LightFloatBand.dbc` или
|
||||
`LightSkybox.dbc` разделит согласованный набор. Некоторые кастомные клиенты
|
||||
используют символы, сортирующиеся после букв. Тогда объедините `stage` с
|
||||
реально последним патчем либо удалите конфликтующие DBC и проверьте цепочку с
|
||||
`--require-winner ANY`. Для обычного patch-Z оставьте строгую проверку.
|
||||
|
||||
Расширенный четырёхполевой `LightSkybox.dbc` предназначен для benilla. Не
|
||||
используйте его с неизменённым оригинальным exe 1.12, ожидающим два поля.
|
||||
|
||||
## Удаление
|
||||
|
||||
Закройте игру. Если patch-Z существовал ранее, восстановите резервную копию:
|
||||
|
||||
```powershell
|
||||
Move-Item -Force "C:\Games\WoW-1.12.1\Data\patch-Z.before-benilla.mpq.bak" "C:\Games\WoW-1.12.1\Data\patch-Z.mpq"
|
||||
```
|
||||
|
||||
Если архив был создан этим комплектом с нуля, удалите только его:
|
||||
|
||||
```powershell
|
||||
Remove-Item -LiteralPath "C:\Games\WoW-1.12.1\Data\patch-Z.mpq"
|
||||
```
|
||||
|
||||
Удаление старого patch-Z уничтожит другое пользовательское содержимое.
|
||||
Проверяйте `archive_base` и сохраняйте резервную копию.
|
||||
|
||||
## Решение проблем
|
||||
|
||||
- **Неверная ревизия pywowlib:** повторите шаг 2; сборщик намеренно не принимает
|
||||
другую ревизию.
|
||||
- **StormLib не найден:** укажите `--stormlib` либо `STORMLIB_PATH`; DLL должна
|
||||
быть 64-битной для 64-битного Python.
|
||||
- **CASC build/config не найден:** укажите корень World of Warcraft с
|
||||
`.build.info` и `Data` либо его дочерний `_classic_beta_`.
|
||||
- **Зашифрованный/отсутствующий блок:** обновите публичный список TACT ключей.
|
||||
Приватные ключи не используются.
|
||||
- **Неожиданная схема или номер клона:** другой аддон уже изменил выигрывающие
|
||||
DBC. Уберите конфликт или выполните осознанное объединение.
|
||||
- **Небо не видно:** включите `zoneSkyboxes`, полностью перезапустите клиент и
|
||||
запустите `verify_client_chain.py` на его настоящем `Data`.
|
||||
- **Ошибка band rows:** запустите `check_bands.py <Data> --stormlib <DLL>`.
|
||||
Правильные ключи: `(P-1)*18+b+1` и `(P-1)*6+b+1`.
|
||||
- **Выходной архив уже существует:** выберите новый `--out` либо используйте
|
||||
`--force` только для намеренной замены кандидата.
|
||||
|
||||
## Авторы и права
|
||||
|
||||
Графика небес WoW Forever/retail принадлежит © Blizzard Entertainment и
|
||||
извлекается пользователем локально из собственной установки. Репозиторий не
|
||||
распространяет графику Blizzard или DBC клиента.
|
||||
|
||||
Метод цветокоррекции основан на `wxl-retail-grading` проекта WarcraftXL автора
|
||||
iThorgrim и используется с указанием авторства. LUT и `MonkeyZoneGrade.dbc` в
|
||||
`data/` созданы проектом benilla и лицензированы MIT OR Apache-2.0.
|
||||
Binary file not shown.
@@ -0,0 +1,13 @@
|
||||
# License
|
||||
|
||||
The files in this directory are generated by `tools/build_grading_data.py` and
|
||||
are original benilla project data. They are licensed under either of:
|
||||
|
||||
- MIT (`../../../LICENSE-MIT`)
|
||||
- Apache License 2.0 (`../../../LICENSE-APACHE`)
|
||||
|
||||
at your option.
|
||||
|
||||
No Blizzard asset or client-derived DBC is stored here. Sky models, textures,
|
||||
and lighting DBC merges are produced only from the user's own installations at
|
||||
build time and are written outside the repository.
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,113 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Build MONKEY post-lane BLP2 LUT strips and MonkeyZoneGrade.dbc.
|
||||
|
||||
Outputs ordinary loose patch-tree files below this directory. Copy the resulting
|
||||
DBFilesClient/ and World/ trees into an isolated Data directory for capture runs.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
import math
|
||||
import struct
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent / "data"
|
||||
LUT_DIR = ROOT / "World" / "LUTs"
|
||||
DBC_DIR = ROOT / "DBFilesClient"
|
||||
|
||||
|
||||
def clamp(value):
|
||||
return max(0.0, min(1.0, value))
|
||||
|
||||
|
||||
def identity(c):
|
||||
return c
|
||||
|
||||
|
||||
def elwynn(c):
|
||||
r, g, b = c
|
||||
return (clamp(r * 1.025 + 0.010 * g), clamp(g * 1.035 + 0.012 * r), clamp(b * 0.975))
|
||||
|
||||
|
||||
def duskwood(c):
|
||||
r, g, b = c
|
||||
y = r * 0.299 + g * 0.587 + b * 0.114
|
||||
r, g, b = (y + (r - y) * 0.78, y + (g - y) * 0.78, y + (b - y) * 0.78)
|
||||
return (clamp(r * 0.94), clamp(g * 0.97), clamp(b * 1.045 + 0.008))
|
||||
|
||||
|
||||
def westfall(c):
|
||||
r, g, b = c
|
||||
return (clamp(r * 1.045 + 0.008 * g), clamp(g * 1.018), clamp(b * 0.94))
|
||||
|
||||
|
||||
def redridge(c):
|
||||
r, g, b = c
|
||||
return (clamp(r * 1.055 + 0.008), clamp(g * 0.982), clamp(b * 0.955))
|
||||
|
||||
|
||||
def write_blp(path, transform):
|
||||
width, height = 1024, 32
|
||||
header_bytes = 148
|
||||
palette_bytes = 256 * 4
|
||||
pixel_offset = header_bytes + palette_bytes
|
||||
pixel_size = width * height * 4
|
||||
header = bytearray(header_bytes)
|
||||
header[0:4] = b"BLP2"
|
||||
struct.pack_into("<I", header, 4, 1) # Direct content
|
||||
header[8:12] = bytes((3, 8, 0, 0)) # Raw3 BGRA8, no mip chain
|
||||
struct.pack_into("<II", header, 12, width, height)
|
||||
struct.pack_into("<I", header, 20, pixel_offset)
|
||||
struct.pack_into("<I", header, 84, pixel_size)
|
||||
|
||||
pixels = bytearray()
|
||||
for g in range(32):
|
||||
for b in range(32):
|
||||
for r in range(32):
|
||||
source = (r / 31.0, g / 31.0, b / 31.0)
|
||||
rr, gg, bb = transform(source)
|
||||
rgba = [round(clamp(v) * 255.0) for v in (rr, gg, bb)]
|
||||
pixels.extend((rgba[2], rgba[1], rgba[0], 255)) # BLP Raw3 is BGRA
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(header + bytes(palette_bytes) + pixels)
|
||||
|
||||
|
||||
def write_dbc(path):
|
||||
# ID, AreaID, DayLUT, NightLUT, Strength. Area IDs are the zone-level AreaTable rows.
|
||||
rows = [
|
||||
(1, 12, "World\\LUTs\\ElwynnWarmGreen.blp", "World\\LUTs\\Identity.blp", 0.34),
|
||||
(2, 10, "World\\LUTs\\Identity.blp", "World\\LUTs\\DuskwoodCold.blp", 0.44),
|
||||
(3, 40, "World\\LUTs\\WestfallGolden.blp", "World\\LUTs\\Identity.blp", 0.38),
|
||||
(4, 44, "World\\LUTs\\RedridgeWarmRed.blp", "World\\LUTs\\Identity.blp", 0.34),
|
||||
]
|
||||
strings = bytearray(b"\0")
|
||||
offsets = {"": 0}
|
||||
|
||||
def string_offset(value):
|
||||
if value not in offsets:
|
||||
offsets[value] = len(strings)
|
||||
strings.extend(value.encode("utf-8") + b"\0")
|
||||
return offsets[value]
|
||||
|
||||
records = bytearray()
|
||||
for row_id, area_id, day, night, strength in rows:
|
||||
records.extend(struct.pack("<IIIIf", row_id, area_id, string_offset(day), string_offset(night), strength))
|
||||
header = b"WDBC" + struct.pack("<IIII", len(rows), 5, 20, len(strings))
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(header + records + strings)
|
||||
|
||||
|
||||
def main():
|
||||
for name, transform in [
|
||||
("Identity.blp", identity),
|
||||
("ElwynnWarmGreen.blp", elwynn),
|
||||
("DuskwoodCold.blp", duskwood),
|
||||
("WestfallGolden.blp", westfall),
|
||||
("RedridgeWarmRed.blp", redridge),
|
||||
]:
|
||||
write_blp(LUT_DIR / name, transform)
|
||||
write_dbc(DBC_DIR / "MonkeyZoneGrade.dbc")
|
||||
print(f"wrote 5 LUTs to {LUT_DIR}")
|
||||
print(f"wrote {DBC_DIR / 'MonkeyZoneGrade.dbc'}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,257 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Build an asset-free optional sky and colour-grading MPQ from user installs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import sys
|
||||
import urllib.request
|
||||
|
||||
from casc import CascReader
|
||||
from dbc_build import REQUIRED, build_merged, check_bands
|
||||
from mpq_tools import Storm, find_stormlib, winning_payloads
|
||||
from sky_convert import convert_all, extract_sources, load_pywowlib, read_listfile
|
||||
from wdc import WDC5
|
||||
from wf_colours import apply_wf_colours
|
||||
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
DATA = HERE.parent / "data"
|
||||
LISTFILE_URL = (
|
||||
"https://github.com/wowdev/wow-listfile/releases/latest/download/"
|
||||
"community-listfile.csv"
|
||||
)
|
||||
TACT_KEYS_URL = "https://raw.githubusercontent.com/wowdev/TACTKeys/master/WoW.txt"
|
||||
|
||||
|
||||
def sha256(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
while True:
|
||||
chunk = stream.read(1024 * 1024)
|
||||
if not chunk:
|
||||
break
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def fetch(url: str, target: Path) -> Path:
|
||||
if target.is_file() and target.stat().st_size:
|
||||
return target
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
partial = target.with_suffix(target.suffix + ".part")
|
||||
with urllib.request.urlopen(url, timeout=120) as response, partial.open("wb") as output:
|
||||
shutil.copyfileobj(response, output, 1024 * 1024)
|
||||
partial.replace(target)
|
||||
return target
|
||||
|
||||
|
||||
def casc_root(install: Path) -> Path:
|
||||
install = Path(install).resolve()
|
||||
if (install / "Data" / "config").is_dir():
|
||||
return install
|
||||
if (install.parent / "Data" / "config").is_dir():
|
||||
return install.parent
|
||||
raise FileNotFoundError("no shared CASC Data/config below %s or its parent" % install)
|
||||
|
||||
|
||||
def select_build_key(install: Path, preferred: str | None = None) -> tuple[str, str]:
|
||||
info = install / ".build.info"
|
||||
if not info.is_file():
|
||||
raise FileNotFoundError(info)
|
||||
with info.open("r", encoding="utf-8-sig", newline="") as stream:
|
||||
reader = csv.reader(stream, delimiter="|")
|
||||
raw_headers = next(reader)
|
||||
headers = [header.split("!", 1)[0] for header in raw_headers]
|
||||
rows = [dict(zip(headers, row)) for row in reader]
|
||||
active = [row for row in rows if row.get("Active") == "1"] or rows
|
||||
products = [preferred] if preferred else []
|
||||
products.extend(name for name in ("wow_classic_beta", "wow") if name != preferred)
|
||||
for wanted in products:
|
||||
for row in active:
|
||||
if row.get("Product") == wanted:
|
||||
key = row.get("Build Key", "").lower()
|
||||
if len(key) == 32:
|
||||
return key, wanted
|
||||
raise RuntimeError("no active WoW Forever/classic-beta or retail build in .build.info")
|
||||
|
||||
|
||||
def reset_owned_directory(path: Path, parent: Path, force: bool) -> None:
|
||||
path = path.resolve()
|
||||
parent = parent.resolve()
|
||||
if path.parent != parent:
|
||||
raise RuntimeError("refusing to reset unexpected path %s" % path)
|
||||
if path.exists():
|
||||
if not force:
|
||||
raise FileExistsError("build directory exists; pass --force: %s" % path)
|
||||
shutil.rmtree(path)
|
||||
path.mkdir(parents=True)
|
||||
|
||||
|
||||
def is_within(path: Path, parent: Path) -> bool:
|
||||
try:
|
||||
path.relative_to(parent)
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def copy_own_grading(stage: Path) -> None:
|
||||
for source in sorted(DATA.rglob("*")):
|
||||
if not source.is_file() or source.name == "LICENSE.md":
|
||||
continue
|
||||
relative = source.relative_to(DATA)
|
||||
target = stage / relative
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(source, target)
|
||||
|
||||
|
||||
def archive_base(client_data: Path, letter: str) -> Path | None:
|
||||
wanted = ("patch-%s.mpq" % letter).casefold()
|
||||
matches = [
|
||||
path for path in client_data.iterdir()
|
||||
if path.is_file() and path.name.casefold() == wanted
|
||||
]
|
||||
if len(matches) > 1:
|
||||
raise RuntimeError("multiple case-insensitive matches for %s" % wanted)
|
||||
return matches[0] if matches else None
|
||||
|
||||
|
||||
def stage_pairs(stage: Path) -> list[tuple[Path, str]]:
|
||||
return [
|
||||
(path, str(path.relative_to(stage)).replace("/", "\\"))
|
||||
for path in sorted(stage.rglob("*"))
|
||||
if path.is_file()
|
||||
]
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--wow-forever", type=Path, help="WoW Forever or retail install root")
|
||||
parser.add_argument("--client-data", type=Path, required=True, help="1.12.1 Data directory")
|
||||
parser.add_argument("--out", type=Path, required=True, help="output directory")
|
||||
parser.add_argument("--grading-only", action="store_true", help="pack only project-owned grading data")
|
||||
parser.add_argument("--wf-colours", action="store_true", help="moderate locally extracted WoW Forever night colours")
|
||||
parser.add_argument("--pywowlib", type=Path, help="pinned external pywowlib checkout")
|
||||
parser.add_argument("--stormlib", type=Path, help="path to StormLib.dll")
|
||||
parser.add_argument("--listfile", type=Path, help="existing community-listfile.csv")
|
||||
parser.add_argument("--tact-keys", type=Path, help="existing public WoW TACT key list")
|
||||
parser.add_argument("--patch-letter", default="Z", help="single patch character (default: Z)")
|
||||
parser.add_argument("--force", action="store_true", help="replace an existing output archive")
|
||||
args = parser.parse_args()
|
||||
if len(args.patch_letter) != 1 or not args.patch_letter.isalpha():
|
||||
parser.error("--patch-letter must be one ASCII letter")
|
||||
if not args.client_data.is_dir():
|
||||
parser.error("--client-data is not a directory")
|
||||
if args.wf_colours and args.grading_only:
|
||||
parser.error("--wf-colours cannot be combined with --grading-only")
|
||||
if not args.grading_only and (not args.wow_forever or not args.pywowlib):
|
||||
parser.error("sky builds require --wow-forever and --pywowlib")
|
||||
return args
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if sys.version_info < (3, 8):
|
||||
raise RuntimeError("Python 3.8 or newer is required")
|
||||
args = parse_args()
|
||||
output = args.out.resolve()
|
||||
client_data = args.client_data.resolve()
|
||||
if is_within(output, client_data):
|
||||
raise ValueError("--out must not be inside the 1.12.1 Data directory")
|
||||
if args.wow_forever:
|
||||
modern_input = casc_root(args.wow_forever)
|
||||
if is_within(output, modern_input):
|
||||
raise ValueError("--out must not be inside the modern WoW installation")
|
||||
output.mkdir(parents=True, exist_ok=True)
|
||||
archive = output / ("patch-%s.MPQ" % args.patch_letter.upper())
|
||||
if archive.exists() and not args.force:
|
||||
raise FileExistsError("output archive exists; pass --force: %s" % archive)
|
||||
stage, work = output / "stage", output / "work"
|
||||
reset_owned_directory(stage, output, args.force)
|
||||
reset_owned_directory(work, output, args.force)
|
||||
copy_own_grading(stage)
|
||||
|
||||
storm = Storm(find_stormlib(args.stormlib))
|
||||
report: dict[str, object] = {"grading_only": args.grading_only}
|
||||
if not args.grading_only:
|
||||
cache = output / "cache"
|
||||
listfile_path = args.listfile or fetch(LISTFILE_URL, cache / "community-listfile.csv")
|
||||
tact_keys = args.tact_keys or fetch(TACT_KEYS_URL, cache / "WoW.txt")
|
||||
install = modern_input
|
||||
requested_name = args.wow_forever.resolve().name.casefold()
|
||||
preferred = {
|
||||
"_classic_beta_": "wow_classic_beta",
|
||||
"_retail_": "wow",
|
||||
}.get(requested_name)
|
||||
build_key, product = select_build_key(install, preferred)
|
||||
reader = CascReader(str(install), build_key, str(listfile_path), str(tact_keys))
|
||||
reader.load_encoding()
|
||||
reader.load_root()
|
||||
modules = load_pywowlib(args.pywowlib)
|
||||
source_root = work / "source"
|
||||
extract_sources(reader, read_listfile(listfile_path), source_root, modules)
|
||||
conversions = convert_all(source_root, stage, listfile_path, modules)
|
||||
|
||||
winners, winner_names = winning_payloads(storm, args.client_data, REQUIRED)
|
||||
wf_changed = 0
|
||||
if args.wf_colours:
|
||||
light_data = WDC5(reader.open_fdid(1375580))
|
||||
winners["LightIntBand.dbc"], wf_changed = apply_wf_colours(
|
||||
winners["LightIntBand.dbc"], light_data
|
||||
)
|
||||
merged, dbc_report = build_merged(winners)
|
||||
dbc_dir = stage / "DBFilesClient"
|
||||
dbc_dir.mkdir(parents=True, exist_ok=True)
|
||||
for name, payload in merged.items():
|
||||
(dbc_dir / name).write_bytes(payload)
|
||||
check_bands(merged["LightIntBand.dbc"], merged["LightFloatBand.dbc"])
|
||||
report.update(
|
||||
{
|
||||
"modern_product": product,
|
||||
"build_key": build_key,
|
||||
"dbc_winners": winner_names,
|
||||
"dbc_merge": dbc_report,
|
||||
"wf_colour_rows_changed": wf_changed,
|
||||
"conversions": conversions,
|
||||
}
|
||||
)
|
||||
|
||||
if archive.exists():
|
||||
archive.unlink()
|
||||
base = archive_base(args.client_data, args.patch_letter)
|
||||
storm.pack(archive, stage_pairs(stage), base_archive=base)
|
||||
|
||||
files = {
|
||||
str(path.relative_to(stage)).replace("/", "\\"): {
|
||||
"bytes": path.stat().st_size,
|
||||
"sha256": sha256(path),
|
||||
}
|
||||
for path in sorted(stage.rglob("*"))
|
||||
if path.is_file()
|
||||
}
|
||||
report.update(
|
||||
{
|
||||
"archive": str(archive),
|
||||
"archive_base": str(base) if base else None,
|
||||
"archive_bytes": archive.stat().st_size,
|
||||
"archive_sha256": sha256(archive),
|
||||
"stage_files": files,
|
||||
}
|
||||
)
|
||||
(output / "build-report.json").write_text(
|
||||
json.dumps(report, indent=2) + "\n", encoding="utf-8"
|
||||
)
|
||||
print(
|
||||
"built %s with %d staged files%s"
|
||||
% (archive, len(files), " (WoW Forever colours)" if args.wf_colours else "")
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,413 @@
|
||||
"""Small, read-only local CASC/TACT reader for the installed WoW builds.
|
||||
|
||||
The implementation deliberately indexes only the latest local v7 .idx file per
|
||||
bucket and opens payloads on demand. It never writes to the CASC installation.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import pathlib
|
||||
import struct
|
||||
import zlib
|
||||
from typing import Dict, Iterable, List, Optional, Tuple
|
||||
|
||||
|
||||
class CascError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class EncryptedBlock(CascError):
|
||||
def __init__(self, key_name: bytes):
|
||||
super().__init__("encrypted BLTE block; missing TACT key %s" % key_name.hex())
|
||||
self.key_name = key_name
|
||||
|
||||
|
||||
def _u24be(data: bytes) -> int:
|
||||
return int.from_bytes(data, "big")
|
||||
|
||||
|
||||
def _rotl32(x: int, n: int) -> int:
|
||||
return ((x << n) | (x >> (32 - n))) & 0xFFFFFFFF
|
||||
|
||||
|
||||
def _salsa20_block(key: bytes, nonce8: bytes, counter: int) -> bytes:
|
||||
# Salsa20/20 with the 64-bit nonce/64-bit block-counter construction.
|
||||
if len(key) == 16:
|
||||
constants = b"expand 16-byte k"
|
||||
key = key + key
|
||||
elif len(key) == 32:
|
||||
constants = b"expand 32-byte k"
|
||||
else:
|
||||
raise CascError("Salsa20 key must contain 16 or 32 bytes")
|
||||
c = struct.unpack("<4I", constants)
|
||||
k = struct.unpack("<8I", key)
|
||||
n0, n1 = struct.unpack("<2I", nonce8)
|
||||
state = [c[0], k[0], k[1], k[2], k[3], c[1], n0, n1,
|
||||
counter & 0xFFFFFFFF, (counter >> 32) & 0xFFFFFFFF,
|
||||
c[2], k[4], k[5], k[6], k[7], c[3]]
|
||||
x = state[:]
|
||||
for _ in range(10):
|
||||
# Column round then row round (Salsa20 double-round).
|
||||
for a, b, c_, d in ((0, 4, 8, 12), (5, 9, 13, 1),
|
||||
(10, 14, 2, 6), (15, 3, 7, 11),
|
||||
(0, 1, 2, 3), (5, 6, 7, 4),
|
||||
(10, 11, 8, 9), (15, 12, 13, 14)):
|
||||
x[b] ^= _rotl32((x[a] + x[d]) & 0xFFFFFFFF, 7)
|
||||
x[c_] ^= _rotl32((x[b] + x[a]) & 0xFFFFFFFF, 9)
|
||||
x[d] ^= _rotl32((x[c_] + x[b]) & 0xFFFFFFFF, 13)
|
||||
x[a] ^= _rotl32((x[d] + x[c_]) & 0xFFFFFFFF, 18)
|
||||
return struct.pack("<16I", *((x[i] + state[i]) & 0xFFFFFFFF for i in range(16)))
|
||||
|
||||
|
||||
def salsa20_xor(data: bytes, key: bytes, nonce8: bytes) -> bytes:
|
||||
out = bytearray(len(data))
|
||||
for pos in range(0, len(data), 64):
|
||||
stream = _salsa20_block(key, nonce8, pos // 64)
|
||||
part = data[pos:pos + 64]
|
||||
for j, value in enumerate(part):
|
||||
out[pos + j] = value ^ stream[j]
|
||||
return bytes(out)
|
||||
|
||||
|
||||
def load_tact_keys(path: pathlib.Path) -> Dict[bytes, bytes]:
|
||||
keys: Dict[bytes, bytes] = {}
|
||||
if not path.exists():
|
||||
return keys
|
||||
for raw in path.read_text(encoding="utf-8", errors="replace").splitlines():
|
||||
line = raw.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
bits = line.replace(";", " ").split()
|
||||
if len(bits) < 2:
|
||||
continue
|
||||
try:
|
||||
name, key = bytes.fromhex(bits[0]), bytes.fromhex(bits[1])
|
||||
except ValueError:
|
||||
continue
|
||||
if len(name) == 8 and len(key) in (16, 32):
|
||||
keys[name] = key
|
||||
return keys
|
||||
|
||||
|
||||
class CascReader:
|
||||
"""Read a product from one local modern Battle.net CASC installation."""
|
||||
|
||||
def __init__(self, install: str, build_key: str,
|
||||
listfile: Optional[str] = None,
|
||||
tact_keys: Optional[str] = None):
|
||||
self.install = pathlib.Path(install)
|
||||
self.data_dir = self.install / "Data" / "data"
|
||||
self.config_dir = self.install / "Data" / "config"
|
||||
self.build_key = build_key.lower()
|
||||
self.config = self._read_config(self.build_key)
|
||||
self.keys = load_tact_keys(pathlib.Path(tact_keys)) if tact_keys else {}
|
||||
self.by_ekey9: Dict[bytes, Tuple[int, int, int]] = {}
|
||||
self._load_local_indices()
|
||||
self.names_by_fdid: Dict[int, str] = {}
|
||||
self.fdids_by_name: Dict[str, int] = {}
|
||||
if listfile:
|
||||
self._load_listfile(pathlib.Path(listfile))
|
||||
self.encoding_map: Dict[bytes, List[bytes]] = {}
|
||||
self.root_map: Dict[int, List[bytes]] = {}
|
||||
self.encoding_header = {}
|
||||
self.root_header = {}
|
||||
|
||||
def _config_path(self, key: str) -> pathlib.Path:
|
||||
return self.config_dir / key[:2] / key[2:4] / key
|
||||
|
||||
def _read_config(self, key: str) -> Dict[str, List[str]]:
|
||||
result: Dict[str, List[str]] = {}
|
||||
for raw in self._config_path(key).read_text(encoding="ascii").splitlines():
|
||||
line = raw.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
name, value = line.split("=", 1)
|
||||
result[name.strip()] = value.strip().split()
|
||||
return result
|
||||
|
||||
def _latest_indices(self) -> Iterable[pathlib.Path]:
|
||||
groups: Dict[str, pathlib.Path] = {}
|
||||
for path in self.data_dir.glob("*.idx"):
|
||||
bucket = path.name[:2].lower()
|
||||
old = groups.get(bucket)
|
||||
# Generation is encoded in the remaining filename hex. The newest
|
||||
# index in each bucket is sufficient because local indices are full.
|
||||
if old is None or int(path.stem[2:], 16) > int(old.stem[2:], 16):
|
||||
groups[bucket] = path
|
||||
if len(groups) != 16:
|
||||
raise CascError("expected 16 local index buckets, found %d" % len(groups))
|
||||
return (groups["%02x" % i] for i in range(16))
|
||||
|
||||
def _load_local_indices(self) -> None:
|
||||
for expected_bucket, path in enumerate(self._latest_indices()):
|
||||
data = path.read_bytes()
|
||||
if len(data) < 40:
|
||||
raise CascError("short local index " + str(path))
|
||||
header_len, = struct.unpack_from("<I", data, 0)
|
||||
version, bucket, extra, size_bytes, offset_bytes, key_bytes, checksum_bytes = \
|
||||
struct.unpack_from("<H6B", data, 8)
|
||||
if header_len != 16 or version != 7 or bucket != expected_bucket:
|
||||
raise CascError("unexpected v7 index header in " + str(path))
|
||||
if (extra, size_bytes, offset_bytes, key_bytes) != (0, 4, 5, 9):
|
||||
raise CascError("unsupported v7 index geometry in " + str(path))
|
||||
entries_len, = struct.unpack_from("<I", data, 32)
|
||||
entry_size = key_bytes + offset_bytes + size_bytes
|
||||
if entries_len % entry_size:
|
||||
raise CascError("misaligned index entries in " + str(path))
|
||||
pos = 40
|
||||
end = pos + entries_len
|
||||
if end > len(data):
|
||||
raise CascError("truncated index entries in " + str(path))
|
||||
while pos < end:
|
||||
key9 = data[pos:pos + key_bytes]
|
||||
packed = int.from_bytes(data[pos + key_bytes:pos + key_bytes + offset_bytes], "big")
|
||||
stored_size = int.from_bytes(
|
||||
data[pos + key_bytes + offset_bytes:pos + entry_size], "little")
|
||||
archive = packed >> 30
|
||||
offset = packed & 0x3FFFFFFF
|
||||
self.by_ekey9[key9] = (archive, offset, stored_size)
|
||||
pos += entry_size
|
||||
|
||||
def _load_listfile(self, path: pathlib.Path) -> None:
|
||||
with path.open("r", encoding="utf-8", errors="replace") as stream:
|
||||
for raw in stream:
|
||||
try:
|
||||
left, name = raw.rstrip("\r\n").split(";", 1)
|
||||
fdid = int(left)
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
normalized = name.replace("\\", "/").lower()
|
||||
self.names_by_fdid[fdid] = name
|
||||
self.fdids_by_name[normalized] = fdid
|
||||
|
||||
def _decode_blte_chunk(self, chunk: bytes, chunk_index: int) -> bytes:
|
||||
if not chunk:
|
||||
return b""
|
||||
mode, payload = chunk[0:1], chunk[1:]
|
||||
if mode == b"N":
|
||||
return payload
|
||||
if mode == b"Z":
|
||||
return zlib.decompress(payload)
|
||||
if mode == b"E":
|
||||
if len(payload) < 3:
|
||||
raise CascError("short encrypted BLTE block")
|
||||
nkey = payload[0]
|
||||
key_name = payload[1:1 + nkey][::-1]
|
||||
at = 1 + nkey
|
||||
niv = payload[at]
|
||||
iv = bytearray(payload[at + 1:at + 1 + niv])
|
||||
at += 1 + niv
|
||||
algorithm = payload[at:at + 1]
|
||||
ciphertext = payload[at + 1:]
|
||||
key = self.keys.get(key_name)
|
||||
if key is None:
|
||||
raise EncryptedBlock(key_name)
|
||||
if algorithm != b"S":
|
||||
raise CascError("unknown BLTE encryption algorithm %r" % algorithm)
|
||||
for i in range(4):
|
||||
if i < len(iv):
|
||||
iv[i] ^= (chunk_index >> (8 * i)) & 0xFF
|
||||
nonce = bytes(iv[:8]).ljust(8, b"\0")
|
||||
clear = salsa20_xor(ciphertext, key, nonce)
|
||||
return self._decode_blte_chunk(clear, chunk_index)
|
||||
raise CascError("unknown BLTE block mode %r" % mode)
|
||||
|
||||
def decode_blte(self, data: bytes) -> bytes:
|
||||
if data[:4] != b"BLTE":
|
||||
raise CascError("payload is not BLTE")
|
||||
header_size, = struct.unpack_from(">I", data, 4)
|
||||
if header_size == 0:
|
||||
return self._decode_blte_chunk(data[8:], 0)
|
||||
if header_size < 12 or header_size > len(data):
|
||||
raise CascError("invalid BLTE header size")
|
||||
flags = data[8]
|
||||
count = _u24be(data[9:12])
|
||||
if flags != 0x0F or header_size < 12 + count * 24:
|
||||
raise CascError("invalid BLTE chunk table")
|
||||
chunks = []
|
||||
pos = 12
|
||||
for _ in range(count):
|
||||
compressed, decompressed = struct.unpack_from(">II", data, pos)
|
||||
digest = data[pos + 8:pos + 24]
|
||||
chunks.append((compressed, decompressed, digest))
|
||||
pos += 24
|
||||
pos = header_size
|
||||
output = []
|
||||
for index, (compressed, decompressed, digest) in enumerate(chunks):
|
||||
raw = data[pos:pos + compressed]
|
||||
if len(raw) != compressed:
|
||||
raise CascError("truncated BLTE chunk")
|
||||
if hashlib.md5(raw).digest() != digest:
|
||||
raise CascError("BLTE chunk checksum mismatch")
|
||||
clear = self._decode_blte_chunk(raw, index)
|
||||
if len(clear) != decompressed:
|
||||
raise CascError("BLTE decompressed-size mismatch")
|
||||
output.append(clear)
|
||||
pos += compressed
|
||||
return b"".join(output)
|
||||
|
||||
def open_ekey(self, ekey: bytes) -> bytes:
|
||||
if isinstance(ekey, str):
|
||||
ekey = bytes.fromhex(ekey)
|
||||
location = self.by_ekey9.get(ekey[:9])
|
||||
if location is None:
|
||||
raise CascError("encoding key is not present locally: " + ekey.hex())
|
||||
archive, offset, stored_size = location
|
||||
data_path = self.data_dir / ("data.%03d" % archive)
|
||||
with data_path.open("rb") as stream:
|
||||
stream.seek(offset)
|
||||
record = stream.read(stored_size)
|
||||
if len(record) != stored_size or len(record) < 30:
|
||||
raise CascError("short local CASC record for " + ekey.hex())
|
||||
# Local data header: reversed 16-byte (zero-extended) EKey, payload size,
|
||||
# flags/checksum fields; the encoded BLTE stream begins at byte 30.
|
||||
if record[30:34] != b"BLTE":
|
||||
raise CascError("local CASC record has no BLTE payload")
|
||||
return self.decode_blte(record[30:])
|
||||
|
||||
def load_encoding(self) -> None:
|
||||
pair = self.config.get("encoding")
|
||||
if not pair or len(pair) < 2:
|
||||
raise CascError("build config has no encoding CKey/EKey pair")
|
||||
data = self.open_ekey(bytes.fromhex(pair[1]))
|
||||
self._parse_encoding(data)
|
||||
|
||||
def _parse_encoding(self, data: bytes) -> None:
|
||||
if data[:2] != b"EN":
|
||||
raise CascError("bad encoding-file signature")
|
||||
# Header fields are big-endian on disk.
|
||||
version, ckey_size, ekey_size = data[2], data[3], data[4]
|
||||
cpage_kib, epage_kib = struct.unpack_from(">HH", data, 5)
|
||||
cpages, epages = struct.unpack_from(">II", data, 9)
|
||||
unknown = data[17]
|
||||
espec_size, = struct.unpack_from(">I", data, 18)
|
||||
self.encoding_header = dict(version=version, ckey_size=ckey_size,
|
||||
ekey_size=ekey_size, cpage_kib=cpage_kib,
|
||||
epage_kib=epage_kib, cpages=cpages,
|
||||
epages=epages, unknown=unknown,
|
||||
espec_size=espec_size,
|
||||
decoded_size=len(data))
|
||||
# The ESpec string block follows the 22-byte header. Each CKey page-index
|
||||
# entry is first-key+MD5; the fixed-size pages immediately follow it.
|
||||
pos = 22 + espec_size
|
||||
cpage_size = cpage_kib * 1024
|
||||
epage_size = epage_kib * 1024
|
||||
cindex_pos = pos
|
||||
cdata_pos = cindex_pos + cpages * (ckey_size + 16)
|
||||
mapping: Dict[bytes, List[bytes]] = {}
|
||||
for page in range(cpages):
|
||||
page_data = data[cdata_pos + page * cpage_size:cdata_pos + (page + 1) * cpage_size]
|
||||
at = 0
|
||||
while at < len(page_data):
|
||||
if at + 2 > len(page_data):
|
||||
break
|
||||
count = int.from_bytes(page_data[at:at + 2], "little")
|
||||
at += 2
|
||||
if count == 0:
|
||||
break
|
||||
need = 4 + ckey_size + count * ekey_size
|
||||
if at + need > len(page_data):
|
||||
raise CascError("truncated encoding CKey page")
|
||||
_decoded_size = int.from_bytes(page_data[at:at + 4], "big")
|
||||
at += 4
|
||||
ckey = page_data[at:at + ckey_size]
|
||||
at += ckey_size
|
||||
ekeys = []
|
||||
for _ in range(count):
|
||||
ekeys.append(page_data[at:at + ekey_size])
|
||||
at += ekey_size
|
||||
mapping[ckey] = ekeys
|
||||
self.encoding_map = mapping
|
||||
|
||||
def open_ckey(self, ckey: bytes) -> bytes:
|
||||
if isinstance(ckey, str):
|
||||
ckey = bytes.fromhex(ckey)
|
||||
if not self.encoding_map:
|
||||
self.load_encoding()
|
||||
ekeys = self.encoding_map.get(ckey)
|
||||
if not ekeys:
|
||||
raise CascError("content key absent from encoding file: " + ckey.hex())
|
||||
errors = []
|
||||
for ekey in ekeys:
|
||||
try:
|
||||
return self.open_ekey(ekey)
|
||||
except CascError as exc:
|
||||
errors.append(str(exc))
|
||||
raise CascError("no local encoding for CKey %s: %s" % (ckey.hex(), "; ".join(errors)))
|
||||
|
||||
def load_root(self) -> None:
|
||||
roots = self.config.get("root")
|
||||
if not roots:
|
||||
raise CascError("build config has no root CKey")
|
||||
data = self.open_ckey(bytes.fromhex(roots[0]))
|
||||
self._parse_root(data)
|
||||
|
||||
def _parse_root(self, data: bytes) -> None:
|
||||
if data[:4] != b"TSFM" or len(data) < 24:
|
||||
raise CascError("unsupported root signature")
|
||||
header_size, version, total_files, named_files, flags = \
|
||||
struct.unpack_from("<5I", data, 4)
|
||||
if header_size != 24 or version != 2:
|
||||
raise CascError("unsupported TSFM root version/header")
|
||||
mapping: Dict[int, List[bytes]] = {}
|
||||
pos = header_size
|
||||
seen_total = 0
|
||||
seen_named = 0
|
||||
blocks = 0
|
||||
while pos < len(data):
|
||||
if pos + 17 > len(data):
|
||||
raise CascError("truncated TSFM block header")
|
||||
count, content_flags, locale_flags = struct.unpack_from("<III", data, pos)
|
||||
block_meta = data[pos + 12:pos + 17]
|
||||
if count == 0 or block_meta[:3] != b"\0\0\0":
|
||||
raise CascError("invalid TSFM block at offset %d" % pos)
|
||||
delta_pos = pos + 17
|
||||
ckey_pos = delta_pos + count * 4
|
||||
hash_pos = ckey_pos + count * 16
|
||||
# In TSFM v2 bit 0x10 of the fourth metadata byte marks entries
|
||||
# without filename hashes. This agrees with the root header's named
|
||||
# count and is also how the manifest stays compact for unnamed FDIDs.
|
||||
has_name_hash = (block_meta[3] & 0x10) == 0
|
||||
end = hash_pos + (count * 8 if has_name_hash else 0)
|
||||
if end > len(data):
|
||||
raise CascError("truncated TSFM block data")
|
||||
fdid = -1
|
||||
for index in range(count):
|
||||
delta, = struct.unpack_from("<I", data, delta_pos + index * 4)
|
||||
fdid += delta + 1
|
||||
ckey = data[ckey_pos + index * 16:ckey_pos + (index + 1) * 16]
|
||||
values = mapping.setdefault(fdid, [])
|
||||
if ckey not in values:
|
||||
values.append(ckey)
|
||||
seen_total += count
|
||||
if has_name_hash:
|
||||
seen_named += count
|
||||
blocks += 1
|
||||
pos = end
|
||||
if seen_total != total_files or seen_named != named_files:
|
||||
raise CascError("TSFM record-count mismatch")
|
||||
self.root_header = dict(header_size=header_size, version=version,
|
||||
total_files=total_files, named_files=named_files,
|
||||
flags=flags, blocks=blocks)
|
||||
self.root_map = mapping
|
||||
|
||||
def open_fdid(self, fdid: int) -> bytes:
|
||||
if not self.root_map:
|
||||
self.load_root()
|
||||
ckeys = self.root_map.get(int(fdid))
|
||||
if not ckeys:
|
||||
raise CascError("FileDataID absent from root: %d" % fdid)
|
||||
errors = []
|
||||
for ckey in ckeys:
|
||||
try:
|
||||
return self.open_ckey(ckey)
|
||||
except CascError as exc:
|
||||
errors.append(str(exc))
|
||||
raise CascError("no readable root entry for FDID %d: %s" % (fdid, "; ".join(errors)))
|
||||
|
||||
def open_name(self, path: str) -> bytes:
|
||||
normalized = path.replace("\\", "/").lower()
|
||||
fdid = self.fdids_by_name.get(normalized)
|
||||
if fdid is None:
|
||||
raise CascError("path absent from listfile: " + path)
|
||||
return self.open_fdid(fdid)
|
||||
@@ -0,0 +1,36 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Check corrected Light band clone row IDs in loose files or an MPQ chain."""
|
||||
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
|
||||
from dbc_build import check_bands
|
||||
from mpq_tools import Storm, copies_for, find_stormlib
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("path", type=Path, help="stage directory or client Data directory")
|
||||
parser.add_argument("--stormlib", type=Path)
|
||||
parser.add_argument("--stage", action="store_true", help="path is a loose stage tree")
|
||||
args = parser.parse_args()
|
||||
if args.stage:
|
||||
root = args.path / "DBFilesClient"
|
||||
integer = (root / "LightIntBand.dbc").read_bytes()
|
||||
floating = (root / "LightFloatBand.dbc").read_bytes()
|
||||
else:
|
||||
storm = Storm(find_stormlib(args.stormlib))
|
||||
payloads = []
|
||||
for name in ("LightIntBand.dbc", "LightFloatBand.dbc"):
|
||||
copies = copies_for(storm, args.path, "DBFilesClient\\" + name)
|
||||
if not copies:
|
||||
raise FileNotFoundError(name)
|
||||
payloads.append(copies[-1][1])
|
||||
integer, floating = payloads
|
||||
check_bands(integer, floating)
|
||||
print("PASS: clone rows use (P-1)*count + band + 1 and match params 23/21")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,200 @@
|
||||
"""Winner-safe lighting DBC merge for the optional sky patch."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import struct
|
||||
from typing import Dict, List, Set, Tuple
|
||||
|
||||
|
||||
REQUIRED = (
|
||||
"Light.dbc",
|
||||
"LightParams.dbc",
|
||||
"LightIntBand.dbc",
|
||||
"LightFloatBand.dbc",
|
||||
"LightSkybox.dbc",
|
||||
)
|
||||
|
||||
|
||||
def parse_dbc(data: bytes, label: str) -> Tuple[List[List[int]], bytes, int]:
|
||||
if len(data) < 20:
|
||||
raise ValueError("truncated DBC: %s" % label)
|
||||
magic, count, fields, record_size, string_size = struct.unpack_from("<4s4I", data)
|
||||
if magic != b"WDBC" or record_size != fields * 4:
|
||||
raise ValueError("invalid WDBC schema: %s" % label)
|
||||
expected = 20 + count * record_size + string_size
|
||||
if len(data) != expected:
|
||||
raise ValueError("invalid WDBC length: %s" % label)
|
||||
records = [
|
||||
list(struct.unpack_from("<%dI" % fields, data, 20 + i * record_size))
|
||||
for i in range(count)
|
||||
]
|
||||
return records, data[20 + count * record_size :], fields
|
||||
|
||||
|
||||
def pack_dbc(records: List[List[int]], strings: bytes, fields: int) -> bytes:
|
||||
if any(len(record) != fields for record in records):
|
||||
raise ValueError("inconsistent DBC record width")
|
||||
header = struct.pack("<4s4I", b"WDBC", len(records), fields, fields * 4, len(strings))
|
||||
body = b"".join(struct.pack("<%dI" % fields, *record) for record in records)
|
||||
return header + body + strings
|
||||
|
||||
|
||||
def indexed(records: List[List[int]], label: str) -> Dict[int, List[int]]:
|
||||
result = {record[0]: record for record in records}
|
||||
if len(result) != len(records):
|
||||
raise ValueError("duplicate row ID in %s" % label)
|
||||
return result
|
||||
|
||||
|
||||
def band_row_id(param: int, band: int, per: int) -> int:
|
||||
return (param - 1) * per + band + 1
|
||||
|
||||
|
||||
def append_string(block: bytearray, value: str) -> int:
|
||||
encoded = value.encode("utf-8") + b"\0"
|
||||
start = 0
|
||||
while True:
|
||||
offset = block.find(encoded, start)
|
||||
if offset < 0:
|
||||
offset = len(block)
|
||||
block.extend(encoded)
|
||||
return offset
|
||||
if offset == 0 or block[offset - 1] == 0:
|
||||
return offset
|
||||
start = offset + 1
|
||||
|
||||
|
||||
def build_merged(winners: Dict[str, bytes]) -> Tuple[Dict[str, bytes], Dict[str, object]]:
|
||||
missing = [name for name in REQUIRED if name not in winners]
|
||||
if missing:
|
||||
raise ValueError("missing winning source DBCs: %s" % ", ".join(missing))
|
||||
|
||||
light, light_strings, light_fields = parse_dbc(winners["Light.dbc"], "Light")
|
||||
params, params_strings, params_fields = parse_dbc(
|
||||
winners["LightParams.dbc"], "LightParams"
|
||||
)
|
||||
ints, int_strings, int_fields = parse_dbc(
|
||||
winners["LightIntBand.dbc"], "LightIntBand"
|
||||
)
|
||||
floats, float_strings, float_fields = parse_dbc(
|
||||
winners["LightFloatBand.dbc"], "LightFloatBand"
|
||||
)
|
||||
skies, sky_strings, sky_fields = parse_dbc(
|
||||
winners["LightSkybox.dbc"], "LightSkybox"
|
||||
)
|
||||
if (light_fields, params_fields, int_fields, float_fields, sky_fields) != (
|
||||
12, 9, 34, 34, 2
|
||||
):
|
||||
raise ValueError("unexpected source lighting schema; use an unmodified 1.12 DBC chain")
|
||||
if any(block != b"\0" for block in (
|
||||
light_strings, params_strings, int_strings, float_strings
|
||||
)):
|
||||
raise ValueError("numeric lighting DBC has a non-empty string block")
|
||||
|
||||
light_by_id = indexed(light, "Light")
|
||||
params_by_id = indexed(params, "LightParams")
|
||||
int_by_id = indexed(ints, "LightIntBand")
|
||||
float_by_id = indexed(floats, "LightFloatBand")
|
||||
|
||||
targets = {20: 1139, 539: 1139, 6: 1140, 65: 1140, 67: 1140}
|
||||
expected = {20: 23, 539: 23, 6: 21, 65: 21, 67: 21}
|
||||
if light_by_id.get(14, [None] * 8)[7] != 16:
|
||||
raise ValueError("Light row 14 is not on expected Swamp of Sorrows param 16")
|
||||
for row_id, new_param in targets.items():
|
||||
if row_id not in light_by_id or light_by_id[row_id][7] != expected[row_id]:
|
||||
raise ValueError("unexpected Light[%d].Clear" % row_id)
|
||||
light_by_id[row_id][7] = new_param
|
||||
|
||||
start_param = max(
|
||||
max(params_by_id),
|
||||
(max(int_by_id) + 17) // 18,
|
||||
(max(float_by_id) + 5) // 6,
|
||||
) + 1
|
||||
if start_param != 1139:
|
||||
raise ValueError(
|
||||
"winner-derived clone start is %d, expected 1139; refusing an unsafe merge"
|
||||
% start_param
|
||||
)
|
||||
|
||||
for source_param, new_param, skybox_id in ((23, 1139, 8), (21, 1140, 9)):
|
||||
clone = params_by_id[source_param][:]
|
||||
clone[0] = new_param
|
||||
clone[2] = skybox_id
|
||||
params.append(clone)
|
||||
for per, source, destination in (
|
||||
(18, int_by_id, ints),
|
||||
(6, float_by_id, floats),
|
||||
):
|
||||
for band in range(per):
|
||||
source_id = band_row_id(source_param, band, per)
|
||||
target_id = band_row_id(new_param, band, per)
|
||||
if source_id not in source or target_id in source:
|
||||
raise ValueError("unsafe band clone %d -> %d" % (source_id, target_id))
|
||||
row = source[source_id][:]
|
||||
row[0] = target_id
|
||||
destination.append(row)
|
||||
|
||||
if params_by_id.get(269, [None, None, None])[2] != 0:
|
||||
raise ValueError("unexpected LightParams[269].LightSkybox")
|
||||
params_by_id[269][2] = 10
|
||||
|
||||
strings = bytearray(sky_strings)
|
||||
extended = [[row[0], row[1], 0, 0] for row in skies]
|
||||
specs = (
|
||||
(8, r"Environments\Stars\10can_volcanicsky01.mdx", 15, ""),
|
||||
(9, r"Environments\Stars\bonewastesskybox.mdx", 4, ""),
|
||||
(
|
||||
10,
|
||||
r"Environments\Stars\8drk_darkshoreelune_sky01.mdx",
|
||||
2,
|
||||
r"Environments\Stars\hyjal_skybox.mdx",
|
||||
),
|
||||
)
|
||||
for row_id, model, flags, celestial in specs:
|
||||
extended.append(
|
||||
[row_id, append_string(strings, model), flags, append_string(strings, celestial)]
|
||||
)
|
||||
|
||||
light.sort(key=lambda row: row[0])
|
||||
params.sort(key=lambda row: row[0])
|
||||
ints.sort(key=lambda row: row[0])
|
||||
floats.sort(key=lambda row: row[0])
|
||||
extended.sort(key=lambda row: row[0])
|
||||
outputs = {
|
||||
"Light.dbc": pack_dbc(light, light_strings, 12),
|
||||
"LightParams.dbc": pack_dbc(params, params_strings, 9),
|
||||
"LightIntBand.dbc": pack_dbc(ints, int_strings, 34),
|
||||
"LightFloatBand.dbc": pack_dbc(floats, float_strings, 34),
|
||||
"LightSkybox.dbc": pack_dbc(extended, bytes(strings), 4),
|
||||
}
|
||||
report = {
|
||||
"source_rows": {
|
||||
name: len(parse_dbc(winners[name], name)[0]) for name in REQUIRED
|
||||
},
|
||||
"output_rows": {
|
||||
name: len(parse_dbc(payload, name)[0]) for name, payload in outputs.items()
|
||||
},
|
||||
"changed_light_rows": sorted(targets),
|
||||
"light_14_clear": 16,
|
||||
"clone_params": [1139, 1140],
|
||||
}
|
||||
return outputs, report
|
||||
|
||||
|
||||
def check_bands(int_payload: bytes, float_payload: bytes) -> None:
|
||||
for label, payload, per in (
|
||||
("LightIntBand", int_payload, 18),
|
||||
("LightFloatBand", float_payload, 6),
|
||||
):
|
||||
records, _, _ = parse_dbc(payload, label)
|
||||
rows = indexed(records, label)
|
||||
for source_param, target_param in ((23, 1139), (21, 1140)):
|
||||
for band in range(per):
|
||||
source_id = band_row_id(source_param, band, per)
|
||||
target_id = band_row_id(target_param, band, per)
|
||||
if target_id not in rows or rows[source_id][1:] != rows[target_id][1:]:
|
||||
raise ValueError("%s clone mismatch at row %d" % (label, target_id))
|
||||
top = band_row_id(1140, per - 1, per)
|
||||
stray = [row_id for row_id in rows if row_id > top]
|
||||
if stray:
|
||||
raise ValueError("%s has rows above param 1140 span" % label)
|
||||
@@ -0,0 +1,243 @@
|
||||
"""Small StormLib wrapper used by the optional-data builder.
|
||||
|
||||
StormLib is loaded from the user's machine. This module never writes to a
|
||||
client installation: candidates are copied or created below the requested
|
||||
output directory and are verified after compaction.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ctypes
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import stat
|
||||
import tempfile
|
||||
from typing import Dict, Iterable, List, Optional, Tuple
|
||||
|
||||
|
||||
MPQ_OPEN_READ_ONLY = 0x00000100
|
||||
MPQ_FILE_COMPRESS = 0x00000200
|
||||
MPQ_FILE_REPLACEEXISTING = 0x80000000
|
||||
MPQ_COMPRESSION_ZLIB = 0x02
|
||||
MAX_FILE_COUNT = 8192
|
||||
|
||||
|
||||
def find_stormlib(explicit: Optional[Path]) -> Path:
|
||||
candidates = []
|
||||
if explicit:
|
||||
candidates.append(Path(explicit))
|
||||
env = os.environ.get("STORMLIB_PATH")
|
||||
if env:
|
||||
candidates.append(Path(env))
|
||||
candidates.extend(
|
||||
[
|
||||
Path.cwd() / "StormLib.dll",
|
||||
Path(__file__).resolve().parent / "StormLib.dll",
|
||||
]
|
||||
)
|
||||
for path in candidates:
|
||||
if path.is_file():
|
||||
return path.resolve()
|
||||
raise FileNotFoundError(
|
||||
"StormLib.dll was not found; pass --stormlib or set STORMLIB_PATH"
|
||||
)
|
||||
|
||||
|
||||
class Storm:
|
||||
def __init__(self, library: Path):
|
||||
if os.name != "nt":
|
||||
raise OSError("this StormLib ctypes wrapper currently supports Windows")
|
||||
self.library = Path(library)
|
||||
self.dll = ctypes.WinDLL(str(self.library), use_last_error=True)
|
||||
handle = ctypes.c_void_p
|
||||
signatures = {
|
||||
"SFileOpenArchive": (
|
||||
ctypes.c_bool,
|
||||
[ctypes.c_wchar_p, ctypes.c_uint32, ctypes.c_uint32, ctypes.POINTER(handle)],
|
||||
),
|
||||
"SFileCreateArchive": (
|
||||
ctypes.c_bool,
|
||||
[ctypes.c_wchar_p, ctypes.c_uint32, ctypes.c_uint32, ctypes.POINTER(handle)],
|
||||
),
|
||||
"SFileSetMaxFileCount": (ctypes.c_bool, [handle, ctypes.c_uint32]),
|
||||
"SFileAddFileEx": (
|
||||
ctypes.c_bool,
|
||||
[handle, ctypes.c_wchar_p, ctypes.c_char_p, ctypes.c_uint32,
|
||||
ctypes.c_uint32, ctypes.c_uint32],
|
||||
),
|
||||
"SFileCompactArchive": (
|
||||
ctypes.c_bool,
|
||||
[handle, ctypes.c_char_p, ctypes.c_bool],
|
||||
),
|
||||
"SFileOpenFileEx": (
|
||||
ctypes.c_bool,
|
||||
[handle, ctypes.c_char_p, ctypes.c_uint32, ctypes.POINTER(handle)],
|
||||
),
|
||||
"SFileGetFileSize": (
|
||||
ctypes.c_uint32,
|
||||
[handle, ctypes.POINTER(ctypes.c_uint32)],
|
||||
),
|
||||
"SFileReadFile": (
|
||||
ctypes.c_bool,
|
||||
[handle, ctypes.c_void_p, ctypes.c_uint32,
|
||||
ctypes.POINTER(ctypes.c_uint32), ctypes.c_void_p],
|
||||
),
|
||||
"SFileCloseFile": (ctypes.c_bool, [handle]),
|
||||
"SFileCloseArchive": (ctypes.c_bool, [handle]),
|
||||
}
|
||||
for name, (result, arguments) in signatures.items():
|
||||
function = getattr(self.dll, name)
|
||||
function.restype = result
|
||||
function.argtypes = arguments
|
||||
|
||||
@staticmethod
|
||||
def fail(message: str) -> None:
|
||||
error = ctypes.get_last_error()
|
||||
raise OSError(error, "%s (Win32 error %d)" % (message, error))
|
||||
|
||||
def open(self, archive: Path, flags: int = 0) -> ctypes.c_void_p:
|
||||
handle = ctypes.c_void_p()
|
||||
if not self.dll.SFileOpenArchive(str(archive), 0, flags, ctypes.byref(handle)):
|
||||
self.fail("SFileOpenArchive failed: %s" % archive)
|
||||
return handle
|
||||
|
||||
def create(self, archive: Path) -> ctypes.c_void_p:
|
||||
handle = ctypes.c_void_p()
|
||||
if not self.dll.SFileCreateArchive(
|
||||
str(archive), 0, MAX_FILE_COUNT, ctypes.byref(handle)
|
||||
):
|
||||
self.fail("SFileCreateArchive failed: %s" % archive)
|
||||
return handle
|
||||
|
||||
def close_archive(self, handle: ctypes.c_void_p) -> None:
|
||||
if handle.value and not self.dll.SFileCloseArchive(handle):
|
||||
self.fail("SFileCloseArchive failed")
|
||||
|
||||
def read(self, archive: Path, member: str) -> Optional[bytes]:
|
||||
archive_handle = self.open(archive, MPQ_OPEN_READ_ONLY)
|
||||
try:
|
||||
file_handle = ctypes.c_void_p()
|
||||
encoded = member.encode("ascii")
|
||||
if not self.dll.SFileOpenFileEx(
|
||||
archive_handle, encoded, 0, ctypes.byref(file_handle)
|
||||
):
|
||||
return None
|
||||
try:
|
||||
high = ctypes.c_uint32(0)
|
||||
size = self.dll.SFileGetFileSize(file_handle, ctypes.byref(high))
|
||||
if size == 0xFFFFFFFF or high.value:
|
||||
self.fail("SFileGetFileSize failed: %s" % member)
|
||||
buffer = ctypes.create_string_buffer(size)
|
||||
received = ctypes.c_uint32(0)
|
||||
if not self.dll.SFileReadFile(
|
||||
file_handle, buffer, size, ctypes.byref(received), None
|
||||
):
|
||||
self.fail("SFileReadFile failed: %s" % member)
|
||||
if received.value != size:
|
||||
raise OSError("short MPQ read for %s" % member)
|
||||
return buffer.raw[:size]
|
||||
finally:
|
||||
if file_handle.value and not self.dll.SFileCloseFile(file_handle):
|
||||
self.fail("SFileCloseFile failed: %s" % member)
|
||||
finally:
|
||||
self.close_archive(archive_handle)
|
||||
|
||||
def pack(
|
||||
self,
|
||||
archive: Path,
|
||||
files: Iterable[Tuple[Path, str]],
|
||||
base_archive: Optional[Path] = None,
|
||||
) -> None:
|
||||
archive.parent.mkdir(parents=True, exist_ok=True)
|
||||
if base_archive:
|
||||
shutil.copy2(base_archive, archive)
|
||||
os.chmod(archive, archive.stat().st_mode | stat.S_IWRITE)
|
||||
handle = self.open(archive)
|
||||
else:
|
||||
handle = self.create(archive)
|
||||
pairs = list(files)
|
||||
listfile_temp = None
|
||||
if not base_archive:
|
||||
listfile_temp = tempfile.TemporaryDirectory(prefix="benilla-listfile-")
|
||||
listfile_path = Path(listfile_temp.name) / "listfile.txt"
|
||||
listfile_path.write_text(
|
||||
"\r\n".join(member for _, member in pairs) + "\r\n(listfile)\r\n",
|
||||
encoding="ascii",
|
||||
)
|
||||
pairs.append((listfile_path, "(listfile)"))
|
||||
try:
|
||||
if not self.dll.SFileSetMaxFileCount(handle, MAX_FILE_COUNT):
|
||||
self.fail("SFileSetMaxFileCount failed")
|
||||
for disk_path, member in pairs:
|
||||
if not self.dll.SFileAddFileEx(
|
||||
handle,
|
||||
str(disk_path),
|
||||
member.encode("ascii"),
|
||||
MPQ_FILE_COMPRESS | MPQ_FILE_REPLACEEXISTING,
|
||||
MPQ_COMPRESSION_ZLIB,
|
||||
MPQ_COMPRESSION_ZLIB,
|
||||
):
|
||||
self.fail("SFileAddFileEx failed: %s" % member)
|
||||
finally:
|
||||
self.close_archive(handle)
|
||||
|
||||
handle = self.open(archive)
|
||||
try:
|
||||
if not self.dll.SFileCompactArchive(handle, None, False):
|
||||
self.fail("SFileCompactArchive failed")
|
||||
finally:
|
||||
self.close_archive(handle)
|
||||
|
||||
for disk_path, member in pairs:
|
||||
actual = self.read(archive, member)
|
||||
if actual != disk_path.read_bytes():
|
||||
raise RuntimeError("packed member failed verification: %s" % member)
|
||||
if listfile_temp is not None:
|
||||
listfile_temp.cleanup()
|
||||
|
||||
|
||||
def mount_key(path: Path) -> Tuple[int, int, str]:
|
||||
name = path.name.casefold()
|
||||
if name == "patch.mpq":
|
||||
return (1, 0, name)
|
||||
numeric = re.fullmatch(r"patch-(\d+)\.mpq", name)
|
||||
if numeric:
|
||||
return (2, int(numeric.group(1)), name)
|
||||
letter = re.fullmatch(r"patch-(.)\.mpq", name)
|
||||
if letter:
|
||||
return (3, ord(letter.group(1)), name)
|
||||
return (0, 0, name)
|
||||
|
||||
|
||||
def archives_in_mount_order(data_dir: Path) -> List[Path]:
|
||||
paths = [
|
||||
path for path in Path(data_dir).iterdir()
|
||||
if path.is_file() and path.suffix.casefold() == ".mpq"
|
||||
]
|
||||
return sorted(paths, key=mount_key)
|
||||
|
||||
|
||||
def copies_for(storm: Storm, data_dir: Path, member: str) -> List[Tuple[Path, bytes]]:
|
||||
result = []
|
||||
for archive in archives_in_mount_order(data_dir):
|
||||
payload = storm.read(archive, member)
|
||||
if payload is not None:
|
||||
result.append((archive, payload))
|
||||
return result
|
||||
|
||||
|
||||
def winning_payloads(
|
||||
storm: Storm, data_dir: Path, names: Iterable[str]
|
||||
) -> Tuple[Dict[str, bytes], Dict[str, str]]:
|
||||
payloads: Dict[str, bytes] = {}
|
||||
winners: Dict[str, str] = {}
|
||||
for name in names:
|
||||
copies = copies_for(storm, data_dir, "DBFilesClient\\" + name)
|
||||
if not copies:
|
||||
raise FileNotFoundError("no mounted copy of DBFilesClient\\%s" % name)
|
||||
archive, payload = copies[-1]
|
||||
payloads[name] = payload
|
||||
winners[name] = archive.name
|
||||
return payloads, winners
|
||||
@@ -0,0 +1,409 @@
|
||||
"""Extract four modern sky models from local CASC and convert them to M2 v256.
|
||||
|
||||
pywowlib is an external dependency. The conversion and v256 compatibility
|
||||
fixes here are original glue; no pywowlib source is redistributed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import Dict, Iterable, List, Tuple
|
||||
|
||||
|
||||
PYWOWLIB_COMMIT = "55276dc5c2195da7fe136638a2a59716622f8c65"
|
||||
MODELS = (
|
||||
("blasted_lands", 130481, "bonewastesskybox.m2"),
|
||||
("mount_hyjal_main", 2322316, "8drk_darkshoreelune_sky01.m2"),
|
||||
("mount_hyjal_celestial", 317347, "hyjal_skybox.m2"),
|
||||
("burning_steppes", 4505901, "10can_volcanicsky01.m2"),
|
||||
)
|
||||
|
||||
|
||||
def load_pywowlib(checkout: Path) -> Dict[str, object]:
|
||||
checkout = Path(checkout).resolve()
|
||||
package = checkout if checkout.name.casefold() == "pywowlib" else checkout / "pywowlib"
|
||||
if not (package / "m2_file.py").is_file():
|
||||
raise FileNotFoundError("--pywowlib must name a pywowlib checkout")
|
||||
try:
|
||||
head = subprocess.check_output(
|
||||
["git", "-C", str(package), "rev-parse", "HEAD"], text=True
|
||||
).strip()
|
||||
except (OSError, subprocess.CalledProcessError) as error:
|
||||
raise RuntimeError("the pywowlib checkout must retain its .git metadata") from error
|
||||
if head != PYWOWLIB_COMMIT:
|
||||
raise RuntimeError(
|
||||
"pywowlib HEAD is %s; expected pinned commit %s" % (head, PYWOWLIB_COMMIT)
|
||||
)
|
||||
sys.path.insert(0, str(package.parent))
|
||||
modules = {
|
||||
"common": importlib.import_module("pywowlib.file_formats.wow_common_types"),
|
||||
"types": importlib.import_module("pywowlib.io_utils.types"),
|
||||
"m2_file": importlib.import_module("pywowlib.m2_file"),
|
||||
"m2_format": importlib.import_module("pywowlib.file_formats.m2_format"),
|
||||
"m2_chunks": importlib.import_module("pywowlib.file_formats.m2_chunks"),
|
||||
"skin_format": importlib.import_module("pywowlib.file_formats.skin_format"),
|
||||
}
|
||||
return modules
|
||||
|
||||
|
||||
def read_listfile(path: Path) -> Dict[int, str]:
|
||||
result = {}
|
||||
with Path(path).open("r", encoding="utf-8", errors="replace") as stream:
|
||||
for line in stream:
|
||||
raw_id, separator, name = line.rstrip("\r\n").partition(";")
|
||||
if separator and raw_id.isdigit():
|
||||
result[int(raw_id)] = name
|
||||
return result
|
||||
|
||||
|
||||
def source_version(path: Path) -> int:
|
||||
data = path.read_bytes()[:16]
|
||||
if data[:4] == b"MD20":
|
||||
return struct.unpack_from("<I", data, 4)[0]
|
||||
if data[:4] == b"MD21" and data[8:12] == b"MD20":
|
||||
return struct.unpack_from("<I", data, 12)[0]
|
||||
raise ValueError("unable to determine M2 version: %s" % path)
|
||||
|
||||
|
||||
def expansion_for(version: int) -> int:
|
||||
versions = {256: 0, 263: 1, 264: 2, 272: 3, 273: 5, 274: 6}
|
||||
if version not in versions:
|
||||
raise ValueError("unsupported M2 version %d" % version)
|
||||
return versions[version]
|
||||
|
||||
|
||||
def extract_sources(reader: object, listfile: Dict[int, str], root: Path, modules: Dict[str, object]) -> None:
|
||||
M2File = modules["m2_file"].M2File
|
||||
manager = modules["common"].M2VersionsManager()
|
||||
for _, file_id, output_name in MODELS:
|
||||
source_dir = root / str(file_id)
|
||||
source_dir.mkdir(parents=True, exist_ok=True)
|
||||
model_path = source_dir / output_name
|
||||
model_path.write_bytes(reader.open_fdid(file_id))
|
||||
version = source_version(model_path)
|
||||
manager.set_m2_version(modules["common"].M2Versions.from_expansion_number(expansion_for(version)))
|
||||
model = M2File(expansion_for(version), filepath=str(model_path))
|
||||
dependencies = model.find_model_dependencies()
|
||||
ids = set(int(value) for value in dependencies.skins)
|
||||
for index, texture in enumerate(model.root.textures):
|
||||
texture_id = int(getattr(texture, "fdid", 0) or 0)
|
||||
if not texture_id and model.txid and index < len(model.txid.texture_ids):
|
||||
texture_id = int(model.txid.texture_ids[index])
|
||||
if texture.type == 0 and not texture.filename.value and texture_id:
|
||||
ids.add(texture_id)
|
||||
for dependency in sorted(ids):
|
||||
game_path = listfile.get(dependency)
|
||||
if not game_path:
|
||||
raise KeyError("FileDataID %d is absent from the community listfile" % dependency)
|
||||
local = source_dir / Path(game_path.replace("\\", "/")).name
|
||||
local.write_bytes(reader.open_fdid(dependency))
|
||||
|
||||
|
||||
def walk_objects(root: object) -> Iterable[object]:
|
||||
seen = set()
|
||||
stack = [root]
|
||||
while stack:
|
||||
value = stack.pop()
|
||||
if id(value) in seen:
|
||||
continue
|
||||
seen.add(id(value))
|
||||
yield value
|
||||
if isinstance(value, (str, bytes, int, float, bool, type(None), tuple)):
|
||||
continue
|
||||
if isinstance(value, list):
|
||||
stack.extend(value)
|
||||
continue
|
||||
values = getattr(value, "values", None)
|
||||
if isinstance(values, list):
|
||||
stack.extend(values)
|
||||
attributes = getattr(value, "__dict__", None)
|
||||
if attributes:
|
||||
stack.extend(attributes.values())
|
||||
|
||||
|
||||
def collect_tracks(root: object, track_base: type) -> List[object]:
|
||||
return [value for value in walk_objects(root) if isinstance(value, track_base)]
|
||||
|
||||
|
||||
def assign_sequence_times(root: object) -> None:
|
||||
timestamp = 0
|
||||
for sequence in root.sequences:
|
||||
duration = int(getattr(sequence, "duration", 0) or 0)
|
||||
sequence.start_timestamp = timestamp
|
||||
sequence.end_timestamp = timestamp + duration
|
||||
timestamp = sequence.end_timestamp + 1000
|
||||
|
||||
|
||||
def denormalize_tracks(root: object, modules: Dict[str, object]) -> None:
|
||||
common, types, fmt = modules["common"], modules["types"], modules["m2_format"]
|
||||
M2Array, M2Range = common.M2Array, fmt.M2Range
|
||||
sequences = list(root.sequences)
|
||||
|
||||
def array(value_type: type, values: List[object]) -> object:
|
||||
result = M2Array(value_type)
|
||||
result.values = list(values)
|
||||
result.n_elements = len(result.values)
|
||||
return result
|
||||
|
||||
for track in collect_tracks(root, fmt.M2TrackBase):
|
||||
timestamp_arrays = track.timestamps.values
|
||||
if not timestamp_arrays or not isinstance(timestamp_arrays[0], M2Array):
|
||||
continue
|
||||
has_values = hasattr(track, "values")
|
||||
value_arrays = track.values.values if has_values else []
|
||||
value_type = value_arrays[0].type if value_arrays else types.uint32
|
||||
flat_times: List[int] = []
|
||||
flat_values: List[object] = []
|
||||
ranges: List[Tuple[int, int]] = []
|
||||
if track.global_sequence >= 0:
|
||||
flat_times = [int(value) for value in timestamp_arrays[0].values]
|
||||
flat_values = list(value_arrays[0].values) if value_arrays else []
|
||||
else:
|
||||
last_value = None
|
||||
for index, sequence in enumerate(sequences):
|
||||
times = timestamp_arrays[index].values if index < len(timestamp_arrays) else []
|
||||
values = value_arrays[index].values if index < len(value_arrays) else []
|
||||
offset = len(flat_times)
|
||||
if times:
|
||||
flat_times.extend(int(value) + sequence.start_timestamp for value in times)
|
||||
if has_values:
|
||||
flat_values.extend(values)
|
||||
last_value = values[-1]
|
||||
ranges.append((offset, offset + len(times) - 1))
|
||||
elif has_values and last_value is not None:
|
||||
flat_times.append(sequence.start_timestamp)
|
||||
flat_values.append(last_value)
|
||||
ranges.append((offset, offset))
|
||||
else:
|
||||
ranges.append((offset, max(0, offset - 1)))
|
||||
ranges.append((0, 0))
|
||||
if has_values and flat_values and isinstance(flat_values[0], fmt.M2CompQuaternion):
|
||||
converted = []
|
||||
for value in flat_values:
|
||||
ww, xx, yy, zz = value.to_quaternion()
|
||||
converted.append((xx, yy, zz, ww))
|
||||
flat_values = converted
|
||||
value_type = types.quat
|
||||
interpolation = M2Array(M2Range)
|
||||
interpolation.values = []
|
||||
for minimum, maximum in ranges:
|
||||
item = M2Range()
|
||||
item.minimum, item.maximum = minimum, maximum
|
||||
interpolation.values.append(item)
|
||||
interpolation.n_elements = len(interpolation.values)
|
||||
track.interpolation_ranges = interpolation
|
||||
track.timestamps = array(types.uint32, flat_times)
|
||||
if has_values:
|
||||
track.values = array(value_type, flat_values)
|
||||
|
||||
|
||||
def stage_sources(shader_id: int, count: int) -> Tuple[int, ...]:
|
||||
shader_id &= 0xFFFF
|
||||
if count <= 1:
|
||||
return (-1,) if shader_id & 0x80 else ((1,) if shader_id & 0x4000 else (0,))
|
||||
first = -1 if shader_id & 0x80 else 0
|
||||
second = -1 if shader_id & 0x8 else (1 if shader_id & 0x4000 else 0)
|
||||
return (first, second)
|
||||
|
||||
|
||||
def find_or_append(table: List[int], values: Tuple[int, ...]) -> int:
|
||||
for index in range(len(table) - len(values) + 1):
|
||||
if tuple(table[index : index + len(values)]) == values:
|
||||
return index
|
||||
table.extend(values)
|
||||
return len(table) - len(values)
|
||||
|
||||
|
||||
def fix_texture_units(model: object) -> None:
|
||||
root = model.root
|
||||
coordinates = [0, 1]
|
||||
weights = list(root.transparency_lookup_table.values)
|
||||
transforms = list(root.texture_transforms_lookup_table.values)
|
||||
textures = list(root.texture_lookup_table.values)
|
||||
pair_cache = {}
|
||||
for skin in model.skins:
|
||||
for unit in skin.texture_units.values:
|
||||
count = int(unit.texture_count)
|
||||
if count not in (1, 2):
|
||||
raise ValueError("unsupported textureCount %d" % count)
|
||||
unit.texture_coord_combo_index = find_or_append(
|
||||
coordinates, stage_sources(unit.shader_id, count)
|
||||
)
|
||||
for stage in range(count):
|
||||
tex = unit.texture_combo_index + stage
|
||||
xform = unit.texture_transform_combo_index + stage
|
||||
if tex >= len(textures) or textures[tex] >= len(root.textures.values):
|
||||
raise ValueError("texture combo is out of range")
|
||||
if xform >= len(transforms) or transforms[xform] >= len(root.texture_transforms.values):
|
||||
raise ValueError("texture transform combo is out of range")
|
||||
weight = unit.texture_weight_combo_index
|
||||
if weight >= len(weights) or weights[weight] >= len(root.texture_weights.values):
|
||||
raise ValueError("texture weight combo is out of range")
|
||||
if count == 2:
|
||||
track = weights[weight]
|
||||
if track not in pair_cache:
|
||||
pair_cache[track] = find_or_append(weights, (track, track))
|
||||
unit.texture_weight_combo_index = pair_cache[track]
|
||||
root.tex_unit_lookup_table.values = coordinates
|
||||
root.tex_unit_lookup_table.n_elements = len(coordinates)
|
||||
root.transparency_lookup_table.values = weights
|
||||
root.transparency_lookup_table.n_elements = len(weights)
|
||||
|
||||
|
||||
def make_classic(model: object, modules: Dict[str, object]) -> None:
|
||||
common, fmt, chunks = modules["common"], modules["m2_format"], modules["m2_chunks"]
|
||||
source_root = model.root
|
||||
assign_sequence_times(source_root)
|
||||
denormalize_tracks(source_root, modules)
|
||||
for sequence in source_root.sequences:
|
||||
if not hasattr(sequence, "blend_time"):
|
||||
sequence.blend_time = max(
|
||||
int(getattr(sequence, "blend_time_in", 0)),
|
||||
int(getattr(sequence, "blend_time_out", 0)),
|
||||
)
|
||||
common.M2VersionsManager().set_m2_version(common.M2Versions.CLASSIC)
|
||||
root = chunks.MD20()
|
||||
for name in list(root.__dict__):
|
||||
if name in source_root.__dict__ and name not in {"magic", "version", "m2_version", "_size"}:
|
||||
setattr(root, name, getattr(source_root, name))
|
||||
model.root = root
|
||||
model.version = common.M2Versions.CLASSIC
|
||||
for value in walk_objects(root):
|
||||
if isinstance(value, fmt.M2TrackBase) and not hasattr(value, "interpolation_ranges"):
|
||||
value.interpolation_ranges = common.M2Array(fmt.M2Range)
|
||||
if hasattr(value, "m2_version"):
|
||||
value.m2_version = common.M2Versions.CLASSIC
|
||||
if value.__class__.__name__ == "M2SkinProfile":
|
||||
value._size = 44
|
||||
for skin in model.skins:
|
||||
for value in walk_objects(skin):
|
||||
if hasattr(value, "m2_version"):
|
||||
value.m2_version = common.M2Versions.CLASSIC
|
||||
if value.__class__.__name__ == "M2SkinProfile":
|
||||
value._size = 44
|
||||
|
||||
|
||||
def write_classic(model: object, path: Path, modules: Dict[str, object]) -> None:
|
||||
skin_class = modules["skin_format"].M2SkinProfile
|
||||
bone_class = modules["m2_format"].M2CompBone
|
||||
types = modules["types"]
|
||||
original_skin_write = skin_class.write
|
||||
original_skin_size = getattr(skin_class, "size", None)
|
||||
original_bone_write = bone_class.write
|
||||
|
||||
def write_embedded(skin: object, stream: object) -> object:
|
||||
position = stream.tell()
|
||||
stream.write(b"\0" * 44)
|
||||
stream.seek(position)
|
||||
skin.vertex_indices.write(stream)
|
||||
skin.triangle_indices.write(stream)
|
||||
skin.bone_indices.write(stream)
|
||||
skin.submeshes.write(stream)
|
||||
skin.texture_units.write(stream)
|
||||
types.uint32.write(stream, skin.bone_count_max)
|
||||
return skin
|
||||
|
||||
def write_classic_bone(bone: object, stream: object) -> object:
|
||||
types.int32.write(stream, bone.key_bone_id)
|
||||
types.uint32.write(stream, bone.flags)
|
||||
types.int16.write(stream, bone.parent_bone)
|
||||
types.uint16.write(stream, bone.submesh_id)
|
||||
bone.translation.write(stream)
|
||||
bone.rotation.write(stream)
|
||||
bone.scale.write(stream)
|
||||
types.vec3D.write(stream, bone.pivot)
|
||||
return bone
|
||||
|
||||
skin_class.write = write_embedded
|
||||
skin_class.size = staticmethod(lambda: 44)
|
||||
bone_class.write = write_classic_bone
|
||||
try:
|
||||
model.root.skin_profiles.values = list(model.skins)
|
||||
model.root.skin_profiles.n_elements = len(model.skins)
|
||||
model.root.num_skin_profiles = len(model.skins)
|
||||
with path.open("wb") as stream:
|
||||
model.root.write(stream)
|
||||
stream.seek(0, 2)
|
||||
padding = (16 - stream.tell() % 16) % 16
|
||||
stream.write(b"\0" * padding)
|
||||
finally:
|
||||
skin_class.write = original_skin_write
|
||||
if original_skin_size is None:
|
||||
delattr(skin_class, "size")
|
||||
else:
|
||||
skin_class.size = original_skin_size
|
||||
bone_class.write = original_bone_write
|
||||
|
||||
|
||||
def convert_all(
|
||||
source_root: Path,
|
||||
output_root: Path,
|
||||
listfile_path: Path,
|
||||
modules: Dict[str, object],
|
||||
) -> List[Dict[str, object]]:
|
||||
listfile = read_listfile(listfile_path)
|
||||
output_root.mkdir(parents=True, exist_ok=True)
|
||||
result = []
|
||||
for key, file_id, output_name in MODELS:
|
||||
source_dir = source_root / str(file_id)
|
||||
source_path = source_dir / output_name
|
||||
version = source_version(source_path)
|
||||
common = modules["common"]
|
||||
common.M2VersionsManager().set_m2_version(
|
||||
common.M2Versions.from_expansion_number(expansion_for(version))
|
||||
)
|
||||
model = modules["m2_file"].M2File(expansion_for(version), filepath=str(source_path))
|
||||
dependencies = model.find_model_dependencies()
|
||||
skins = []
|
||||
for dependency in dependencies.skins:
|
||||
name = listfile.get(int(dependency))
|
||||
if not name:
|
||||
raise KeyError("skin FileDataID %d missing from listfile" % dependency)
|
||||
skins.append(str(source_dir / Path(name.replace("\\", "/")).name))
|
||||
model.read_additional_files(skins, {})
|
||||
textures = []
|
||||
for index, texture in enumerate(model.root.textures):
|
||||
file_id_value = int(getattr(texture, "fdid", 0) or 0)
|
||||
if not file_id_value and model.txid and index < len(model.txid.texture_ids):
|
||||
file_id_value = int(model.txid.texture_ids[index])
|
||||
if texture.type == 0 and not texture.filename.value:
|
||||
game_path = listfile.get(file_id_value)
|
||||
if not game_path:
|
||||
raise KeyError("texture FileDataID %d missing from listfile" % file_id_value)
|
||||
texture.filename.value = game_path.replace("/", "\\")
|
||||
local = source_dir / Path(game_path.replace("\\", "/")).name
|
||||
target = output_root.joinpath(*game_path.replace("\\", "/").split("/"))
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(local, target)
|
||||
textures.append(str(target.relative_to(output_root)).replace("/", "\\"))
|
||||
model.root.global_flags &= 0x17
|
||||
if hasattr(model.root, "texture_combiner_combos"):
|
||||
model.root.texture_combiner_combos.values = []
|
||||
model.root.texture_combiner_combos.n_elements = 0
|
||||
for material in model.root.materials:
|
||||
if int(material.blending_mode) > 6:
|
||||
material.blending_mode = 2
|
||||
fix_texture_units(model)
|
||||
make_classic(model, modules)
|
||||
target = output_root / "Environments" / "Stars" / output_name
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
write_classic(model, target, modules)
|
||||
result.append(
|
||||
{
|
||||
"key": key,
|
||||
"source_file_data_id": file_id,
|
||||
"source_version": version,
|
||||
"output": str(target),
|
||||
"output_bytes": target.stat().st_size,
|
||||
"skins_embedded": len(model.skins),
|
||||
"textures": textures,
|
||||
}
|
||||
)
|
||||
return result
|
||||
@@ -0,0 +1,111 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Verify every staged payload is the mounted winner in a candidate Data chain."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
|
||||
from dbc_build import check_bands, indexed, parse_dbc
|
||||
from mpq_tools import Storm, copies_for, find_stormlib
|
||||
|
||||
|
||||
def digest(data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def semantic_checks(payloads: dict[str, bytes]) -> None:
|
||||
light, _, fields = parse_dbc(payloads["DBFilesClient\\Light.dbc"], "Light")
|
||||
if fields != 12:
|
||||
raise ValueError("Light.dbc is not 12 fields")
|
||||
light_rows = indexed(light, "Light")
|
||||
expected = {20: 1139, 539: 1139, 6: 1140, 65: 1140, 67: 1140, 14: 16}
|
||||
for row_id, clear in expected.items():
|
||||
if light_rows[row_id][7] != clear:
|
||||
raise ValueError("Light[%d].Clear is not %d" % (row_id, clear))
|
||||
params, _, fields = parse_dbc(
|
||||
payloads["DBFilesClient\\LightParams.dbc"], "LightParams"
|
||||
)
|
||||
if fields != 9:
|
||||
raise ValueError("LightParams.dbc is not 9 fields")
|
||||
param_rows = indexed(params, "LightParams")
|
||||
for row_id, skybox in ((1139, 8), (1140, 9), (269, 10)):
|
||||
if param_rows[row_id][2] != skybox:
|
||||
raise ValueError("LightParams[%d] skybox mismatch" % row_id)
|
||||
if 1141 in param_rows:
|
||||
raise ValueError("stray LightParams 1141")
|
||||
check_bands(
|
||||
payloads["DBFilesClient\\LightIntBand.dbc"],
|
||||
payloads["DBFilesClient\\LightFloatBand.dbc"],
|
||||
)
|
||||
skies, strings, fields = parse_dbc(
|
||||
payloads["DBFilesClient\\LightSkybox.dbc"], "LightSkybox"
|
||||
)
|
||||
if fields != 4 or not strings:
|
||||
raise ValueError("LightSkybox.dbc is not the extended four-field layout")
|
||||
sky_rows = indexed(skies, "LightSkybox")
|
||||
for row_id in (8, 9, 10):
|
||||
if row_id not in sky_rows:
|
||||
raise ValueError("LightSkybox row %d is missing" % row_id)
|
||||
grade, _, fields = parse_dbc(
|
||||
payloads["DBFilesClient\\MonkeyZoneGrade.dbc"], "MonkeyZoneGrade"
|
||||
)
|
||||
if fields != 5 or len(grade) != 4:
|
||||
raise ValueError("MonkeyZoneGrade must have four five-field rows")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("data_dir", type=Path)
|
||||
parser.add_argument("--stage", type=Path, required=True)
|
||||
parser.add_argument("--stormlib", type=Path)
|
||||
parser.add_argument("--require-winner", default="patch-Z.mpq")
|
||||
args = parser.parse_args()
|
||||
storm = Storm(find_stormlib(args.stormlib))
|
||||
expected = {
|
||||
str(path.relative_to(args.stage)).replace("/", "\\"): path.read_bytes()
|
||||
for path in sorted(args.stage.rglob("*"))
|
||||
if path.is_file()
|
||||
}
|
||||
mounted = {}
|
||||
failures = []
|
||||
for member, wanted in expected.items():
|
||||
copies = copies_for(storm, args.data_dir, member)
|
||||
if not copies:
|
||||
failures.append("missing %s" % member)
|
||||
continue
|
||||
archive, actual = copies[-1]
|
||||
mounted[member] = actual
|
||||
if args.require_winner.casefold() != "any" and (
|
||||
archive.name.casefold() != args.require_winner.casefold()
|
||||
):
|
||||
failures.append("%s wins from %s" % (member, archive.name))
|
||||
if actual != wanted:
|
||||
failures.append(
|
||||
"%s differs: %s != %s" % (member, digest(actual), digest(wanted))
|
||||
)
|
||||
required = {
|
||||
"DBFilesClient\\Light.dbc",
|
||||
"DBFilesClient\\LightParams.dbc",
|
||||
"DBFilesClient\\LightIntBand.dbc",
|
||||
"DBFilesClient\\LightFloatBand.dbc",
|
||||
"DBFilesClient\\LightSkybox.dbc",
|
||||
"DBFilesClient\\MonkeyZoneGrade.dbc",
|
||||
}
|
||||
if required <= set(mounted):
|
||||
try:
|
||||
semantic_checks(mounted)
|
||||
except (KeyError, ValueError) as error:
|
||||
failures.append("semantic check: %s" % error)
|
||||
if failures:
|
||||
print("FAIL")
|
||||
for failure in failures:
|
||||
print(" " + failure)
|
||||
return 1
|
||||
print("PASS: %d staged payloads are mounted winners" % len(expected))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,171 @@
|
||||
"""Minimal WDC5 reader for dense modern WoW DB2 tables."""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import struct
|
||||
|
||||
|
||||
class WdcError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def _bits_le(data: bytes, bit_offset: int, bit_count: int) -> int:
|
||||
if bit_count == 0:
|
||||
return 0
|
||||
start = bit_offset // 8
|
||||
shift = bit_offset & 7
|
||||
count = (shift + bit_count + 7) // 8
|
||||
value = int.from_bytes(data[start:start + count], "little") >> shift
|
||||
return value & ((1 << bit_count) - 1)
|
||||
|
||||
|
||||
class WDC5:
|
||||
HEADER_OFFSET = 0x88
|
||||
|
||||
def __init__(self, data: bytes):
|
||||
self.data = data
|
||||
if data[:4] != b"WDC5" or len(data) < self.HEADER_OFFSET + 68:
|
||||
raise WdcError("not a supported WDC5 table")
|
||||
o = self.HEADER_OFFSET
|
||||
(self.record_count, self.field_count, self.record_size,
|
||||
self.string_table_size, self.table_hash, self.layout_hash,
|
||||
self.min_id, self.max_id, self.locale) = struct.unpack_from("<9I", data, o)
|
||||
self.flags, self.id_index = struct.unpack_from("<HH", data, o + 36)
|
||||
(self.total_field_count, self.bitpacked_data_offset,
|
||||
self.lookup_column_count, self.field_storage_info_size,
|
||||
self.common_data_size, self.pallet_data_size,
|
||||
self.section_count) = struct.unpack_from("<7I", data, o + 40)
|
||||
if self.section_count != 1:
|
||||
raise WdcError("only single-section WDC5 tables are supported")
|
||||
at = o + 68
|
||||
section = struct.unpack_from("<Q8I", data, at)
|
||||
(self.tact_key_hash, self.file_offset, section_records,
|
||||
self.section_string_size, self.offset_records_end,
|
||||
self.id_list_size, self.relationship_data_size,
|
||||
self.offset_map_id_count, self.copy_table_count) = section
|
||||
if section_records != self.record_count:
|
||||
raise WdcError("multi-section record count mismatch")
|
||||
at += 40
|
||||
self.field_structures = [struct.unpack_from("<hH", data, at + i * 4)
|
||||
for i in range(self.total_field_count)]
|
||||
at += self.total_field_count * 4
|
||||
self.storage = [struct.unpack_from("<HH5I", data, at + i * 24)
|
||||
for i in range(self.total_field_count)]
|
||||
at += self.field_storage_info_size
|
||||
self.pallet = data[at:at + self.pallet_data_size]
|
||||
at += self.pallet_data_size
|
||||
self.common = data[at:at + self.common_data_size]
|
||||
at += self.common_data_size
|
||||
if at != self.file_offset:
|
||||
raise WdcError("WDC5 metadata/file-offset mismatch")
|
||||
self.records_offset = self.file_offset
|
||||
self.strings_offset = self.records_offset + self.record_count * self.record_size
|
||||
after_strings = self.strings_offset + self.section_string_size
|
||||
self.ids_offset = after_strings
|
||||
after_ids = self.ids_offset + self.id_list_size
|
||||
self.copy_offset = after_ids + self.relationship_data_size
|
||||
|
||||
poff = coff = 0
|
||||
self.aux_offsets = []
|
||||
for _field_offset, _field_size, additional, kind, _a, _b, _c in self.storage:
|
||||
if kind in (3, 4):
|
||||
self.aux_offsets.append((poff, None))
|
||||
poff += additional
|
||||
elif kind == 2:
|
||||
self.aux_offsets.append((None, coff))
|
||||
coff += additional
|
||||
else:
|
||||
self.aux_offsets.append((None, None))
|
||||
if poff != self.pallet_data_size or coff != self.common_data_size:
|
||||
raise WdcError("WDC5 auxiliary-data size mismatch")
|
||||
|
||||
if self.id_list_size:
|
||||
if self.id_list_size != self.record_count * 4:
|
||||
raise WdcError("unexpected WDC5 ID-list size")
|
||||
self.ids = list(struct.unpack_from("<%dI" % self.record_count,
|
||||
data, self.ids_offset))
|
||||
else:
|
||||
self.ids = [None] * self.record_count
|
||||
self.rows = {}
|
||||
self.row_ids = []
|
||||
for index in range(self.record_count):
|
||||
recpos = self.records_offset + index * self.record_size
|
||||
record = data[recpos:recpos + self.record_size]
|
||||
values = [self._value(index, record, i) for i in range(self.total_field_count)]
|
||||
row_id = self.ids[index]
|
||||
if row_id is None:
|
||||
row_id = values[self.id_index]
|
||||
self.rows[row_id] = {"id": row_id, "values": values,
|
||||
"record_offset": recpos, "copy_of": None}
|
||||
self.row_ids.append(row_id)
|
||||
self.relationship_by_index = {}
|
||||
if self.relationship_data_size:
|
||||
relpos = after_ids
|
||||
if self.relationship_data_size < 12:
|
||||
raise WdcError("short WDC5 relationship block")
|
||||
count, self.relationship_min_id, self.relationship_max_id = \
|
||||
struct.unpack_from("<3I", data, relpos)
|
||||
if self.relationship_data_size != 12 + count * 8:
|
||||
raise WdcError("unexpected WDC5 relationship-block size")
|
||||
for i in range(count):
|
||||
foreign_id, record_index = struct.unpack_from("<2I", data, relpos + 12 + i * 8)
|
||||
if record_index >= self.record_count:
|
||||
raise WdcError("WDC5 relationship record index is out of bounds")
|
||||
self.relationship_by_index[record_index] = foreign_id
|
||||
self.rows[self.row_ids[record_index]]["parent_id"] = foreign_id
|
||||
for i in range(self.copy_table_count):
|
||||
new_id, source_id = struct.unpack_from("<II", data, self.copy_offset + i * 8)
|
||||
source = self.rows.get(source_id)
|
||||
if source is None:
|
||||
raise WdcError("copy-table source ID is absent")
|
||||
row = copy.deepcopy(source)
|
||||
row["id"] = new_id
|
||||
row["copy_of"] = source_id
|
||||
self.rows[new_id] = row
|
||||
|
||||
def _value(self, row_index: int, record: bytes, field: int):
|
||||
bit_offset, bit_size, additional, kind, a, b, c = self.storage[field]
|
||||
raw = _bits_le(record, bit_offset, bit_size)
|
||||
if kind in (0, 1):
|
||||
return raw
|
||||
if kind == 5:
|
||||
if bit_size and raw & (1 << (bit_size - 1)):
|
||||
raw -= 1 << bit_size
|
||||
return raw
|
||||
if kind == 2:
|
||||
default = a
|
||||
_, coff = self.aux_offsets[field]
|
||||
block = self.common[coff:coff + additional]
|
||||
row_id = self.ids[row_index]
|
||||
for pos in range(0, len(block), 8):
|
||||
key, value = struct.unpack_from("<II", block, pos)
|
||||
if key == row_id:
|
||||
return value
|
||||
return default
|
||||
if kind == 3:
|
||||
poff, _ = self.aux_offsets[field]
|
||||
if raw * 4 + 4 > additional:
|
||||
raise WdcError("palette index out of bounds")
|
||||
return struct.unpack_from("<I", self.pallet, poff + raw * 4)[0]
|
||||
if kind == 4:
|
||||
width = c
|
||||
poff, _ = self.aux_offsets[field]
|
||||
base = poff + raw * width * 4
|
||||
if base + width * 4 > poff + additional:
|
||||
raise WdcError("palette-array index out of bounds")
|
||||
return struct.unpack_from("<%dI" % width, self.pallet, base)
|
||||
raise WdcError("unsupported WDC5 storage type %d" % kind)
|
||||
|
||||
def string(self, row, field: int) -> str:
|
||||
relative = row["values"][field]
|
||||
start = row["record_offset"] + relative
|
||||
if start < self.strings_offset or start >= self.strings_offset + self.section_string_size:
|
||||
return ""
|
||||
end = self.data.find(b"\0", start, self.strings_offset + self.section_string_size)
|
||||
if end < 0:
|
||||
end = self.strings_offset + self.section_string_size
|
||||
return self.data[start:end].decode("utf-8", errors="replace")
|
||||
|
||||
@staticmethod
|
||||
def as_float(value: int) -> float:
|
||||
return struct.unpack("<f", struct.pack("<I", value & 0xFFFFFFFF))[0]
|
||||
@@ -0,0 +1,218 @@
|
||||
"""Moderated WoW Forever LightData colour import.
|
||||
|
||||
The modern table is read directly from the user's CASC installation. Only
|
||||
colour tracks are imported; fog distances, shadow opacity, and modern-only fog
|
||||
fields remain untouched.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import bisect
|
||||
import math
|
||||
import struct
|
||||
from typing import Dict, Iterable, List, Tuple
|
||||
|
||||
from dbc_build import pack_dbc, parse_dbc
|
||||
|
||||
|
||||
DAY = 2880
|
||||
ANCHORS = (0, 720, 1440, 2160)
|
||||
TARGET_PARAMS = (
|
||||
2, 3, 4, 10, 12, 13, 14, 15, 17, 21, 22, 23,
|
||||
24, 25, 192, 193, 206, 213, 269, 271, 454, 456, 494, 497,
|
||||
)
|
||||
DUSKWOOD_LOCAL = (14, 15, 494)
|
||||
BLENDED_BANDS = tuple(index for index in range(18) if index != 8)
|
||||
|
||||
|
||||
def unpack_band(row: List[int]) -> Tuple[List[int], List[int]]:
|
||||
count = row[1]
|
||||
if count > 16:
|
||||
raise ValueError("LightIntBand row has more than 16 keys")
|
||||
return row[2 : 2 + count], row[18 : 18 + count]
|
||||
|
||||
|
||||
def make_band(record_id: int, times: List[int], values: List[int]) -> List[int]:
|
||||
if len(times) != len(values) or len(times) > 16:
|
||||
raise ValueError("invalid band key arrays")
|
||||
return [record_id, len(times)] + times + [0] * (16 - len(times)) + values + [0] * (
|
||||
16 - len(values)
|
||||
)
|
||||
|
||||
|
||||
def rgb_tuple(value: int) -> Tuple[float, float, float]:
|
||||
return (
|
||||
((value >> 16) & 255) / 255.0,
|
||||
((value >> 8) & 255) / 255.0,
|
||||
(value & 255) / 255.0,
|
||||
)
|
||||
|
||||
|
||||
def rgb_int(rgb: Iterable[float]) -> int:
|
||||
channels = [max(0, min(255, round(value * 255.0))) for value in rgb]
|
||||
return (channels[0] << 16) | (channels[1] << 8) | channels[2]
|
||||
|
||||
|
||||
def sample(row: List[int], time: int) -> int:
|
||||
times, values = unpack_band(row)
|
||||
if not times:
|
||||
return 0
|
||||
if len(times) == 1:
|
||||
return values[0] & 0xFFFFFF
|
||||
time %= DAY
|
||||
if time < times[0] or time >= times[-1]:
|
||||
left_t, right_t = times[-1], times[0] + DAY
|
||||
left_v, right_v = values[-1], values[0]
|
||||
query = time + DAY if time < times[0] else time
|
||||
else:
|
||||
right = bisect.bisect_right(times, time)
|
||||
left_t, right_t = times[right - 1], times[right]
|
||||
left_v, right_v = values[right - 1], values[right]
|
||||
query = time
|
||||
amount = (query - left_t) / (right_t - left_t)
|
||||
return rgb_int(
|
||||
a + (b - a) * amount for a, b in zip(rgb_tuple(left_v), rgb_tuple(right_v))
|
||||
)
|
||||
|
||||
|
||||
def srgb_to_linear(value: float) -> float:
|
||||
return value / 12.92 if value <= 0.04045 else ((value + 0.055) / 1.055) ** 2.4
|
||||
|
||||
|
||||
def linear_to_srgb(value: float) -> float:
|
||||
value = max(0.0, value)
|
||||
return 12.92 * value if value <= 0.0031308 else 1.055 * value ** (1 / 2.4) - 0.055
|
||||
|
||||
|
||||
def rgb_to_oklab(rgb: Tuple[float, float, float]) -> Tuple[float, float, float]:
|
||||
red, green, blue = (srgb_to_linear(value) for value in rgb)
|
||||
ll = 0.4122214708 * red + 0.5363325363 * green + 0.0514459929 * blue
|
||||
mm = 0.2119034982 * red + 0.6806995451 * green + 0.1073969566 * blue
|
||||
ss = 0.0883024619 * red + 0.2817188376 * green + 0.6299787005 * blue
|
||||
ll, mm, ss = (
|
||||
math.copysign(abs(ll) ** (1 / 3), ll),
|
||||
math.copysign(abs(mm) ** (1 / 3), mm),
|
||||
math.copysign(abs(ss) ** (1 / 3), ss),
|
||||
)
|
||||
return (
|
||||
0.2104542553 * ll + 0.7936177850 * mm - 0.0040720468 * ss,
|
||||
1.9779984951 * ll - 2.4285922050 * mm + 0.4505937099 * ss,
|
||||
0.0259040371 * ll + 0.7827717662 * mm - 0.8086757660 * ss,
|
||||
)
|
||||
|
||||
|
||||
def oklab_to_rgb(lab: Tuple[float, float, float]) -> Tuple[float, float, float]:
|
||||
light, aa, bb = lab
|
||||
ll = light + 0.3963377774 * aa + 0.2158037573 * bb
|
||||
mm = light - 0.1055613458 * aa - 0.0638541728 * bb
|
||||
ss = light - 0.0894841775 * aa - 1.2914855480 * bb
|
||||
ll, mm, ss = ll ** 3, mm ** 3, ss ** 3
|
||||
red = 4.0767416621 * ll - 3.3077115913 * mm + 0.2309699292 * ss
|
||||
green = -1.2684380046 * ll + 2.6097574011 * mm - 0.3413193965 * ss
|
||||
blue = -0.0041960863 * ll - 0.7034186147 * mm + 1.7076147010 * ss
|
||||
return tuple(
|
||||
max(0.0, min(1.0, linear_to_srgb(value))) for value in (red, green, blue)
|
||||
)
|
||||
|
||||
|
||||
def chroma_hue(lab: Tuple[float, float, float]) -> Tuple[float, float]:
|
||||
_, aa, bb = lab
|
||||
return math.hypot(aa, bb), math.degrees(math.atan2(bb, aa)) % 360.0
|
||||
|
||||
|
||||
def clamp_duskwood(
|
||||
base: Tuple[float, float, float], mixed: Tuple[float, float, float]
|
||||
) -> Tuple[float, float, float]:
|
||||
base_chroma, base_hue = chroma_hue(base)
|
||||
mixed_chroma, mixed_hue = chroma_hue(mixed)
|
||||
mixed_chroma = min(mixed_chroma, base_chroma + 0.015)
|
||||
if mixed_chroma < 1e-8:
|
||||
return (mixed[0], 0.0, 0.0)
|
||||
if base_chroma >= 0.006:
|
||||
shift = (mixed_hue - base_hue + 180.0) % 360.0 - 180.0
|
||||
hue = base_hue + max(-12.0, min(12.0, shift))
|
||||
else:
|
||||
hue = 155.0
|
||||
radians = math.radians(hue)
|
||||
return (
|
||||
mixed[0],
|
||||
mixed_chroma * math.cos(radians),
|
||||
mixed_chroma * math.sin(radians),
|
||||
)
|
||||
|
||||
|
||||
def blend(base_value: int, wf_value: int, amount: float, duskwood: bool) -> int:
|
||||
if base_value == wf_value or amount <= 0.0:
|
||||
return base_value
|
||||
base = rgb_to_oklab(rgb_tuple(base_value))
|
||||
modern = rgb_to_oklab(rgb_tuple(wf_value))
|
||||
mixed = tuple(a + (b - a) * amount for a, b in zip(base, modern))
|
||||
if duskwood:
|
||||
mixed = clamp_duskwood(base, mixed)
|
||||
return rgb_int(oklab_to_rgb(mixed))
|
||||
|
||||
|
||||
def strength(time: int, night: float, day: float) -> float:
|
||||
day_weight = (1.0 - math.cos(2.0 * math.pi * (time % DAY) / DAY)) * 0.5
|
||||
return night + (day - night) * day_weight
|
||||
|
||||
|
||||
def modern_field_for_band(band: int) -> int:
|
||||
if band <= 7:
|
||||
return 3 + band
|
||||
if band == 8:
|
||||
return 20
|
||||
return band + 2
|
||||
|
||||
|
||||
def apply_wf_colours(base_payload: bytes, light_data: object) -> Tuple[bytes, int]:
|
||||
records, strings, fields = parse_dbc(base_payload, "LightIntBand")
|
||||
if fields != 34 or strings != b"\0":
|
||||
raise ValueError("unexpected LightIntBand layout")
|
||||
by_id = {row[0]: row for row in records}
|
||||
modern: Dict[int, List[object]] = {}
|
||||
for row in light_data.rows.values():
|
||||
values = row["values"]
|
||||
param = row.get("parent_id", values[1])
|
||||
if param in TARGET_PARAMS:
|
||||
modern.setdefault(param, []).append(row)
|
||||
missing = [param for param in TARGET_PARAMS if not modern.get(param)]
|
||||
if missing:
|
||||
raise ValueError("modern LightData is missing target params: %s" % missing)
|
||||
|
||||
changed = 0
|
||||
for param in TARGET_PARAMS:
|
||||
rows = sorted(
|
||||
modern[param], key=lambda row: (row["values"][2] & 0xFFFF, row["id"])
|
||||
)
|
||||
if len(rows) > 16:
|
||||
step = (len(rows) - 1) / 15.0
|
||||
rows = [rows[round(index * step)] for index in range(16)]
|
||||
for band in BLENDED_BANDS:
|
||||
record_id = (param - 1) * 18 + band + 1
|
||||
if record_id not in by_id:
|
||||
raise ValueError("missing LightIntBand row %d" % record_id)
|
||||
base_row = by_id[record_id]
|
||||
modern_times = [row["values"][2] & 0xFFFF for row in rows]
|
||||
field = modern_field_for_band(band)
|
||||
modern_values = [row["values"][field] & 0xFFFFFF for row in rows]
|
||||
modern_row = make_band(record_id, modern_times, modern_values)
|
||||
base_times, _ = unpack_band(base_row)
|
||||
times = sorted(set(base_times) | set(modern_times) | set(ANCHORS))
|
||||
if len(times) > 16:
|
||||
raise ValueError("moderated row %d needs more than 16 keys" % record_id)
|
||||
duskwood = param in DUSKWOOD_LOCAL
|
||||
night, day = ((0.30, 0.15) if duskwood else (0.50, 0.25))
|
||||
values = [
|
||||
blend(
|
||||
sample(base_row, time),
|
||||
sample(modern_row, time),
|
||||
strength(time, night, day),
|
||||
duskwood,
|
||||
)
|
||||
for time in times
|
||||
]
|
||||
if any(sample(base_row, time) != value for time, value in zip(times, values)):
|
||||
by_id[record_id][:] = make_band(record_id, times, values)
|
||||
changed += 1
|
||||
return pack_dbc(records, strings, fields), changed
|
||||
@@ -68,6 +68,8 @@ at any IP or hostname, appending the auth port if yours is remapped
|
||||
(`WOW_HOST=play.example.com:5000`). Credentials go in at the login screen, or set `WOW_USER` /
|
||||
`WOW_PASS` to skip it.
|
||||
|
||||
Optional, locally built skybox and colour-grading data: [`Optional/sky-and-grading/`](Optional/sky-and-grading/README.md).
|
||||
|
||||
## Contributing
|
||||
|
||||
Issues and pull requests are open. [`docs/CONTRIBUTING.md`](docs/CONTRIBUTING.md) says what gets
|
||||
|
||||
Reference in New Issue
Block a user