Files
PaintMyKI5/scanner/scan_ki5_workshop.py
T

1165 lines
43 KiB
Python

"""Scan Project Zomboid Workshop mods for KI5 vehicle skin colors."""
from __future__ import annotations
import argparse
import colorsys
import json
import math
import os
import re
import sys
import tempfile
import warnings
from collections import Counter
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from functools import lru_cache
from pathlib import Path
from typing import Iterable, Sequence, TextIO
from PIL import Image, ImageChops, UnidentifiedImageError
DEFAULT_WORKSHOP_ROOT = Path(r"D:\SteamLibrary\steamapps\workshop\content\108600")
DEFAULT_GAME_ROOT = Path(r"D:\SteamLibrary\steamapps\common\ProjectZomboid")
DEFAULT_OUTPUT_PATH = (
Path(__file__).resolve().parent.parent
/ "42.20/media/lua/shared/PaintMyKI5/PaintMyKI5VehiclePaintData.lua"
)
DEPENDENCY_SPLIT = re.compile(r"[,;\s]+")
MODULE_PATTERN = re.compile(r"(?im)^\s*module\s+([\w.-]+)\s*(?=\{)")
VEHICLE_PATTERN = re.compile(r"(?im)^\s*vehicle\s+([\w.-]+)\s*(?=\{)")
SKIN_PATTERN = re.compile(r"(?im)^\s*skin(?:\s+[\w.-]+)?\s*(?=\{)")
TEXTURE_PATTERN = re.compile(r"(?im)^\s*texture\s*=\s*([^,\r\n}]+)")
DRIVABLE_PATTERN = re.compile(
r"(?im)^\s*(?:engineForce|engineLoudness|engineQuality|engineRPMType|maxSpeed)\s*="
)
ITEM_PATTERN = re.compile(r"(?im)^\s*item\s+([\w.-]+)\s*(?=\{)")
PROPERTY_PATTERN = re.compile(r"(?im)(?:^|[,{\r\n])\s*([\w.-]+)\s*=\s*([^,\r\n}]+)")
PAINT_MENU_PATTERN = re.compile(
r'paint\s*=\s*["\']([\w.-]+)["\'][^}\r\n]*'
r'color\s*=\s*\{\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s*,'
r'\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s*,'
r'\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s*\}',
flags=re.IGNORECASE,
)
MAX_TEXT_BYTES = 8 * 1024 * 1024
MAX_IMAGE_PIXELS = 4096 * 4096
MAX_DISCOVERED_FILES = 20_000
MAX_TREE_ENTRIES = 500_000
MAX_UNIQUE_COLORS = 262_144
MAX_TEXTURE_RESULTS = 20_000
MAX_PAINT_CANS = 256
MAX_VEHICLES = 20_000
MAX_SKINS_PER_VEHICLE = 2_000
MAX_WARNINGS = 20_000
MAX_METADATA_CHARS = 2_048
MAX_MANIFEST_BYTES = 64 * 1024 * 1024
MAX_SCAN_DECODED_PIXELS = 3_000_000_000
MAX_VARIANT_MASK_PIXELS = 268_435_456
MAX_VARIANT_MASK_SKINS = 64
CHROMATIC_SATURATION_MIN = 0.10
CHROMATIC_RGB_RANGE_MIN = 8 / 255
MAX_CHROMATIC_HUE_DISTANCE = 0.16
VARIANT_RGB_RANGE_MIN = 16
MIN_VARIANT_MASK_COVERAGE = 0.20
class ScanFileError(ValueError):
"""Raised when an untrusted Workshop file exceeds scanner safety limits."""
@dataclass(frozen=True)
class ModCandidate:
workshop_id: str
mod_id: str
name: str
mod_root: Path
content_root: Path
@dataclass(frozen=True)
class SkinDefinition:
skin_index: int
texture_reference: str
@dataclass(frozen=True)
class VehicleDefinition:
vehicle_id: str
skins: tuple[SkinDefinition, ...]
@property
def texture_references(self) -> tuple[str, ...]:
return tuple(skin.texture_reference for skin in self.skins)
@dataclass(frozen=True)
class PaintCan:
item_id: str
display_name: str
rgb: tuple[float, float, float]
use_delta: float
@dataclass(frozen=True)
class PaintPercentage:
paint_can: PaintCan
percentage: float
@dataclass(frozen=True)
class TextureResult:
workshop_id: str
mod_id: str
mod_name: str
vehicle_id: str
skin_index: int
texture_reference: str
texture_path: Path
paints: tuple[PaintPercentage, ...]
@dataclass(frozen=True)
class ScanReport:
mods_scanned: int
cars_found: int
textures: tuple[TextureResult, ...]
warnings: tuple[str, ...]
def paint_bucket_uses(paint_cans: Sequence[PaintCan]) -> int:
"""Return the full-bucket capacity shared by the discovered B42 paints."""
if not paint_cans:
raise ValueError("The B42 paint palette is empty")
use_deltas = {Decimal(str(paint.use_delta)) for paint in paint_cans}
if len(use_deltas) != 1:
raise ValueError("B42 paint cans do not share one bucket capacity")
use_delta = use_deltas.pop()
if not use_delta.is_finite() or use_delta <= 0:
raise ValueError("B42 paint use delta must be finite and positive")
capacity = Decimal("1") / use_delta
rounded_capacity = capacity.to_integral_value(rounding=ROUND_HALF_UP)
if capacity != rounded_capacity or rounded_capacity <= 0:
raise ValueError("B42 paint use delta does not produce whole bucket uses")
return int(rounded_capacity)
def allocate_paint_uses(
paints: Sequence[PaintPercentage],
bucket_uses: int,
) -> list[tuple[PaintPercentage, float]]:
"""Spread one bucket's uses across every represented paint color."""
if bucket_uses <= 0:
raise ValueError("Paint bucket capacity must be positive")
combined: dict[str, tuple[PaintPercentage, Decimal]] = {}
for paint in paints:
percentage = Decimal(str(paint.percentage))
if not percentage.is_finite() or percentage < 0:
raise ValueError("Paint percentages must be finite and non-negative")
previous = combined.get(paint.paint_can.item_id)
total = percentage + (previous[1] if previous else Decimal("0"))
combined[paint.paint_can.item_id] = (paint, total)
percentage_total = sum((entry[1] for entry in combined.values()), Decimal("0"))
if percentage_total <= 0:
return []
allocations = [
(
entry,
(percentage * Decimal(bucket_uses) / percentage_total).quantize(
Decimal("0.01"), rounding=ROUND_HALF_UP
),
)
for entry, percentage in combined.values()
]
difference = Decimal(bucket_uses) - sum(
(uses for _entry, uses in allocations), Decimal("0")
)
if difference:
target_index = min(
range(len(allocations)),
key=lambda index: (
-Decimal(str(allocations[index][0].percentage)),
allocations[index][0].paint_can.item_id,
),
)
entry, uses = allocations[target_index]
allocations[target_index] = (entry, uses + difference)
return [
(entry, float(uses))
for entry, uses in sorted(
allocations,
key=lambda allocation: (
-allocation[1],
allocation[0].paint_can.item_id,
),
)
if uses > 0
]
def _read_text(path: Path) -> str:
with path.open("rb") as stream:
contents = stream.read(MAX_TEXT_BYTES + 1)
if len(contents) > MAX_TEXT_BYTES:
raise ScanFileError(f"Text file exceeds {MAX_TEXT_BYTES} bytes: {path}")
return contents.decode("utf-8-sig", errors="replace")
def _bounded_files(
root: Path,
*,
name: str | None = None,
suffix: str | None = None,
) -> Iterable[Path]:
visited_entries = 0
yielded_files = 0
pending_directories = [root]
while pending_directories:
directory = pending_directories.pop()
try:
entries = os.scandir(directory)
except OSError:
continue
with entries:
for entry in entries:
visited_entries += 1
if visited_entries > MAX_TREE_ENTRIES:
raise ValueError(
f"File traversal exceeded the {MAX_TREE_ENTRIES}-entry safety limit"
)
try:
if entry.is_dir(follow_symlinks=False):
pending_directories.append(Path(entry.path))
continue
if not entry.is_file(follow_symlinks=False):
continue
except OSError:
continue
file_name = entry.name
if name is not None and file_name.casefold() != name.casefold():
continue
if suffix is not None and not file_name.casefold().endswith(suffix.casefold()):
continue
yielded_files += 1
if yielded_files > MAX_DISCOVERED_FILES:
raise ValueError(
f"File discovery exceeded the {MAX_DISCOVERED_FILES}-file safety limit"
)
yield Path(entry.path)
def _parse_mod_info(path: Path) -> dict[str, str]:
fields: dict[str, str] = {}
for raw_line in _read_text(path).splitlines():
line = raw_line.strip()
if not line or line.startswith(("#", ";", "//")) or "=" not in line:
continue
key, value = line.split("=", 1)
normalized_key = key.strip().casefold()
normalized_value = value.strip().strip("'\"")
if len(normalized_key) > 128 or len(normalized_value) > MAX_METADATA_CHARS:
raise ScanFileError(f"Oversized mod.info field: {path}")
fields[normalized_key] = normalized_value
return fields
def _requires_damnlib(fields: dict[str, str]) -> bool:
dependencies = fields.get("require", "")
tokens = (
token.strip("\\/'\"").casefold()
for token in DEPENDENCY_SPLIT.split(dependencies)
if token
)
return "damnlib" in tokens
def _find_mod_root(info_path: Path) -> tuple[Path, str] | None:
parts = info_path.parts
mod_indexes = [index for index, part in enumerate(parts) if part.casefold() == "mods"]
if not mod_indexes:
return None
mods_index = mod_indexes[-1]
if mods_index == 0 or mods_index + 1 >= len(parts):
return None
return Path(*parts[: mods_index + 2]), parts[mods_index - 1]
def _version_key(path: Path) -> tuple[int, ...] | None:
if not re.fullmatch(r"42(?:\.\d+)*", path.name, flags=re.IGNORECASE):
return None
return tuple(int(part) for part in path.name.split("."))
def _select_content_root(mod_root: Path) -> Path:
resolved_mod_root = mod_root.resolve(strict=True)
version_roots = [
child
for child in mod_root.iterdir()
if child.is_dir()
and _version_key(child) is not None
and not child.is_symlink()
and child.resolve(strict=True).is_relative_to(resolved_mod_root)
and (child / "media" / "scripts" / "vehicles").is_dir()
]
if version_roots:
return max(version_roots, key=lambda path: _version_key(path) or ())
return mod_root
def discover_mods(workshop_root: Path) -> list[ModCandidate]:
"""Return unique DamnLib mods, selecting their newest installed B42 overlay."""
workshop_root = workshop_root.expanduser()
if not workshop_root.is_dir():
raise ValueError(f"Workshop root does not exist or is not a directory: {workshop_root}")
grouped: dict[Path, tuple[str, Path]] = {}
for info_path in _bounded_files(workshop_root, name="mod.info"):
try:
fields = _parse_mod_info(info_path)
except (OSError, ScanFileError):
continue
if not _requires_damnlib(fields):
continue
location = _find_mod_root(info_path)
if location is None:
continue
mod_root, workshop_id = location
previous = grouped.get(mod_root)
fallback_info = (
info_path
if previous is None or len(info_path.parts) < len(previous[1].parts)
else previous[1]
)
grouped[mod_root] = (workshop_id, fallback_info)
candidates: list[ModCandidate] = []
for mod_root, (workshop_id, fallback_info) in grouped.items():
try:
content_root = _select_content_root(mod_root)
except OSError:
continue
selected_info = content_root / "mod.info"
if not selected_info.is_file():
selected_info = fallback_info
try:
fields = _parse_mod_info(selected_info)
except (OSError, ScanFileError):
continue
if not _requires_damnlib(fields):
continue
candidates.append(
ModCandidate(
workshop_id=workshop_id,
mod_id=fields.get("id", mod_root.name),
name=fields.get("name", fields.get("id", mod_root.name)),
mod_root=mod_root,
content_root=content_root,
)
)
return sorted(candidates, key=lambda mod: (mod.workshop_id.casefold(), mod.mod_id.casefold()))
def _without_comments(text: str) -> str:
text = re.sub(r"/\*.*?\*/", "", text, flags=re.DOTALL)
return re.sub(r"//.*$", "", text, flags=re.MULTILINE)
def _extract_braced_block(text: str, opening_brace: int) -> str | None:
if opening_brace < 0 or opening_brace >= len(text) or text[opening_brace] != "{":
return None
depth = 0
for index in range(opening_brace, len(text)):
if text[index] == "{":
depth += 1
elif text[index] == "}":
depth -= 1
if depth == 0:
return text[opening_brace + 1 : index]
return None
def _property_map(block: str) -> dict[str, str]:
return {
match.group(1).casefold(): match.group(2).strip().strip("'\"")
for match in PROPERTY_PATTERN.finditer(block)
}
def _parse_paint_item_ids(item_script: Path) -> dict[str, float]:
text = _without_comments(_read_text(item_script))
paint_items: dict[str, float] = {}
for item_match in ITEM_PATTERN.finditer(text):
opening_brace = text.find("{", item_match.end())
block = _extract_braced_block(text, opening_brace)
if block is None:
continue
properties = _property_map(block)
tags = {
tag.strip().casefold()
for tag in re.split(r"[,;\s]+", properties.get("tags", ""))
if tag.strip()
}
if properties.get("itemtype", "").casefold() != "base:drainable":
continue
if properties.get("pourtype", "").casefold() != "bucket":
continue
if properties.get("replaceondeplete", "").casefold() != "base.paintbucketempty":
continue
if "base:paint" not in tags:
continue
try:
use_delta = float(properties["usedelta"])
except (KeyError, ValueError):
continue
if not 0 < use_delta <= 1:
continue
module_name = _module_before(text, item_match.start())
short_id = item_match.group(1)
item_id = short_id if "." in short_id else f"{module_name}.{short_id}"
paint_items[item_id] = use_delta
return paint_items
def _parse_paint_menu(menu_path: Path) -> dict[str, tuple[float, float, float]]:
text = _without_comments(_read_text(menu_path))
colors: dict[str, tuple[float, float, float]] = {}
for match in PAINT_MENU_PATTERN.finditer(text):
short_id = match.group(1)
item_id = short_id if "." in short_id else f"Base.{short_id}"
rgb = tuple(float(match.group(index)) for index in range(2, 5))
if any(not math.isfinite(channel) or not 0 <= channel <= 1 for channel in rgb):
raise ValueError(f"Invalid B42 paint RGB for {item_id}: {rgb}")
if item_id in colors and colors[item_id] != rgb:
raise ValueError(f"Conflicting B42 paint RGB values for {item_id}")
colors[item_id] = rgb
return colors
def discover_paint_cans(game_root: Path) -> tuple[PaintCan, ...]:
"""Load full B42 paint buckets and canonical RGB values from the game files."""
game_root = game_root.expanduser()
if not game_root.is_dir():
raise ValueError(f"Game root does not exist or is not a directory: {game_root}")
item_script = game_root / "media/scripts/generated/items/drainable.txt"
menu_path = game_root / "media/lua/shared/BuildingObjects/ISPaintMenu.lua"
names_path = game_root / "media/lua/shared/Translate/EN/ItemName.json"
missing = [path for path in (item_script, menu_path) if not path.is_file()]
if missing:
raise ValueError(f"Missing B42 paint data: {missing[0]}")
try:
paint_items = _parse_paint_item_ids(item_script)
menu_colors = _parse_paint_menu(menu_path)
names = json.loads(_read_text(names_path)) if names_path.is_file() else {}
except (OSError, ScanFileError, json.JSONDecodeError) as error:
raise ValueError(f"Could not read B42 paint data: {error}") from error
if not isinstance(names, dict) or not all(
isinstance(key, str) and isinstance(value, str) for key, value in names.items()
):
raise ValueError(f"Invalid B42 item-name data: {names_path}")
cans = tuple(
PaintCan(
item_id=item_id,
display_name=str(names.get(item_id, item_id.removeprefix("Base.Paint") or item_id)),
rgb=menu_colors[item_id],
use_delta=use_delta,
)
for item_id, use_delta in sorted(paint_items.items())
if item_id in menu_colors
)
if not cans:
raise ValueError("No complete B42 paint cans were found in the game data")
missing_colors = sorted(set(paint_items) - set(menu_colors))
if missing_colors:
raise ValueError(f"B42 paint cans are missing RGB values: {', '.join(missing_colors)}")
if len(cans) > MAX_PAINT_CANS:
raise ValueError(f"B42 paint palette exceeds the {MAX_PAINT_CANS}-can safety limit")
return cans
def _module_before(text: str, position: int) -> str:
modules = [match.group(1) for match in MODULE_PATTERN.finditer(text, 0, position)]
return modules[-1] if modules else "Base"
def _vehicle_skins(vehicle_block: str) -> tuple[SkinDefinition, ...]:
skins: list[SkinDefinition] = []
for skin_match in SKIN_PATTERN.finditer(vehicle_block):
opening_brace = vehicle_block.find("{", skin_match.end())
skin_block = _extract_braced_block(vehicle_block, opening_brace)
if skin_block is None:
continue
texture_match = TEXTURE_PATTERN.search(skin_block)
if texture_match:
reference = texture_match.group(1).strip().strip("'\"")
if reference:
if len(reference) > MAX_METADATA_CHARS:
raise ValueError("Vehicle texture reference exceeds the safety limit")
skins.append(SkinDefinition(len(skins), reference))
if len(skins) > MAX_SKINS_PER_VEHICLE:
raise ValueError(
f"Vehicle exceeds the {MAX_SKINS_PER_VEHICLE}-skin safety limit"
)
return tuple(skins)
def parse_vehicle_scripts(scripts_root: Path) -> list[VehicleDefinition]:
"""Parse drivable vehicle IDs and skin texture references from PZ scripts."""
if not scripts_root.is_dir():
return []
vehicles: list[VehicleDefinition] = []
script_paths = sorted(
_bounded_files(scripts_root, suffix=".txt"),
key=lambda path: str(path).casefold(),
)
for script_path in script_paths:
try:
text = _without_comments(_read_text(script_path))
except (OSError, ScanFileError):
continue
for vehicle_match in VEHICLE_PATTERN.finditer(text):
short_vehicle_id = vehicle_match.group(1)
if short_vehicle_id.casefold().startswith("trailer"):
continue
opening_brace = text.find("{", vehicle_match.end())
block = _extract_braced_block(text, opening_brace)
if block is None or not DRIVABLE_PATTERN.search(block):
continue
skins = _vehicle_skins(block)
if not skins:
continue
module_name = _module_before(text, vehicle_match.start())
vehicles.append(
VehicleDefinition(
vehicle_id=f"{module_name}.{short_vehicle_id}",
skins=skins,
)
)
if len(vehicles) > MAX_VEHICLES:
raise ValueError(f"Script scan exceeds the {MAX_VEHICLES}-vehicle safety limit")
unique = {(vehicle.vehicle_id, vehicle.skins): vehicle for vehicle in vehicles}
return sorted(unique.values(), key=lambda vehicle: vehicle.vehicle_id.casefold())
def _case_insensitive_file(root: Path, relative_path: Path) -> Path | None:
current = root
for part in relative_path.parts:
if part in ("", ".", ".."):
return None
direct = current / part
if direct.exists():
current = direct
continue
try:
match = next(
(child for child in current.iterdir() if child.name.casefold() == part.casefold()),
None,
)
except OSError:
return None
if match is None:
return None
current = match
if not current.is_file():
return None
try:
resolved_root = root.resolve(strict=True)
resolved_file = current.resolve(strict=True)
except OSError:
return None
return resolved_file if resolved_file.is_relative_to(resolved_root) else None
def resolve_texture(mod: ModCandidate, reference: str) -> Path | None:
normalized = reference.replace("\\", "/").strip().lstrip("/")
if ":" in normalized:
return None
if not normalized.casefold().endswith(".png"):
normalized += ".png"
relative_path = Path(*normalized.split("/"))
if relative_path.is_absolute() or relative_path.drive or relative_path.anchor:
return None
texture_roots = (
mod.content_root / "media" / "textures",
mod.mod_root / "common" / "media" / "textures",
mod.mod_root / "media" / "textures",
)
for texture_root in texture_roots:
resolved = _case_insensitive_file(texture_root, relative_path)
if resolved is not None:
return resolved
return None
def _srgb_to_lab(rgb: tuple[float, float, float]) -> tuple[float, float, float]:
linear = tuple(
channel / 12.92
if channel <= 0.04045
else ((channel + 0.055) / 1.055) ** 2.4
for channel in rgb
)
red, green, blue = linear
x = (0.4124564 * red + 0.3575761 * green + 0.1804375 * blue) / 0.95047
y = 0.2126729 * red + 0.7151522 * green + 0.0721750 * blue
z = (0.0193339 * red + 0.1191920 * green + 0.9503041 * blue) / 1.08883
def transform(channel: float) -> float:
return channel ** (1 / 3) if channel > 216 / 24389 else (24389 / 27 * channel + 16) / 116
x_value, y_value, z_value = transform(x), transform(y), transform(z)
return 116 * y_value - 16, 500 * (x_value - y_value), 200 * (y_value - z_value)
@lru_cache(maxsize=131_072)
def _nearest_paint(
red: int,
green: int,
blue: int,
palette_colors: tuple[
tuple[PaintCan, tuple[float, float, float], tuple[float, float, float]], ...
],
preserve_chromatic_hue: bool,
) -> PaintCan:
pixel_rgb = red / 255, green / 255, blue / 255
pixel_hsv = colorsys.rgb_to_hsv(*pixel_rgb)
if (
preserve_chromatic_hue
and pixel_hsv[1] >= CHROMATIC_SATURATION_MIN
and max(pixel_rgb) - min(pixel_rgb) >= CHROMATIC_RGB_RANGE_MIN
):
chromatic = [
entry for entry in palette_colors if entry[2][1] >= CHROMATIC_SATURATION_MIN
]
if chromatic:
def chromatic_distance(
entry: tuple[
PaintCan,
tuple[float, float, float],
tuple[float, float, float],
],
) -> tuple[float, str]:
hue_distance = abs(pixel_hsv[0] - entry[2][0])
hue_distance = min(hue_distance, 1 - hue_distance)
return (
(hue_distance * 3) ** 2
+ (pixel_hsv[1] - entry[2][1]) ** 2
+ 0.02 * (pixel_hsv[2] - entry[2][2]) ** 2,
entry[0].item_id,
)
nearest_chromatic = min(chromatic, key=chromatic_distance)
hue_distance = abs(pixel_hsv[0] - nearest_chromatic[2][0])
if min(hue_distance, 1 - hue_distance) <= MAX_CHROMATIC_HUE_DISTANCE:
return nearest_chromatic[0]
pixel_lab = _srgb_to_lab(pixel_rgb)
return min(
palette_colors,
key=lambda entry: (
sum((pixel_lab[index] - entry[1][index]) ** 2 for index in range(3)),
entry[0].item_id,
),
)[0]
def _read_rgba_png(texture_path: Path) -> Image.Image:
try:
with warnings.catch_warnings():
warnings.simplefilter("error", Image.DecompressionBombWarning)
image = Image.open(texture_path, formats=("PNG",))
with image:
if image.format != "PNG":
raise UnidentifiedImageError(f"Not a PNG image: {texture_path}")
width, height = image.size
if width <= 0 or height <= 0 or width * height > MAX_IMAGE_PIXELS:
raise ScanFileError(
f"Image exceeds the {MAX_IMAGE_PIXELS}-pixel safety limit: {texture_path}"
)
return image.convert("RGBA")
except ValueError as error:
raise ScanFileError(f"Could not decode PNG {texture_path}: {error}") from error
def _png_pixel_count(texture_path: Path) -> int:
try:
with warnings.catch_warnings():
warnings.simplefilter("error", Image.DecompressionBombWarning)
with Image.open(texture_path, formats=("PNG",)) as image:
if image.format != "PNG":
raise UnidentifiedImageError(f"Not a PNG image: {texture_path}")
width, height = image.size
except ValueError as error:
raise ScanFileError(f"Could not decode PNG {texture_path}: {error}") from error
pixels = width * height
if width <= 0 or height <= 0 or pixels > MAX_IMAGE_PIXELS:
raise ScanFileError(
f"Image exceeds the {MAX_IMAGE_PIXELS}-pixel safety limit: {texture_path}"
)
return pixels
def _add_decoded_pixels(current: int, additional: int) -> int:
total = current + additional
if additional < 0 or total > MAX_SCAN_DECODED_PIXELS:
raise ScanFileError(
f"Scan exceeds the {MAX_SCAN_DECODED_PIXELS}-decoded-pixel safety limit"
)
return total
def build_variant_mask(texture_paths: Sequence[Path]) -> bytes | None:
"""Select shell pixels whose RGB changes between sibling skin textures."""
unique_paths = tuple(dict.fromkeys(texture_paths))
if len(unique_paths) < 2 or len(unique_paths) > MAX_VARIANT_MASK_SKINS:
return None
pixel_counts = tuple(_png_pixel_count(path) for path in unique_paths)
if sum(pixel_counts) > MAX_VARIANT_MASK_PIXELS:
return None
first = _read_rgba_png(unique_paths[0])
size = first.size
darkest = first.convert("RGB")
lightest = darkest.copy()
shared_opacity = first.getchannel("A")
first.close()
for path in unique_paths[1:]:
image = _read_rgba_png(path)
if image.size != size:
image.close()
return None
rgb = image.convert("RGB")
darkest = ImageChops.darker(darkest, rgb)
lightest = ImageChops.lighter(lightest, rgb)
shared_opacity = ImageChops.darker(shared_opacity, image.getchannel("A"))
image.close()
shared_alpha = shared_opacity.tobytes()
if not any(shared_alpha):
return None
difference = ImageChops.difference(lightest, darkest)
channels = difference.split()
maximum_difference = ImageChops.lighter(ImageChops.lighter(channels[0], channels[1]), channels[2])
mask = maximum_difference.point(
lambda value: 255 if value >= VARIANT_RGB_RANGE_MIN else 0
)
visible_in_every_skin = shared_opacity.point(lambda value: 255 if value else 0)
mask = ImageChops.multiply(mask, visible_in_every_skin)
mask_bytes = mask.tobytes()
visible_pixels = visible_in_every_skin.tobytes().count(255)
masked_pixels = mask_bytes.count(255)
if not visible_pixels or masked_pixels / visible_pixels < MIN_VARIANT_MASK_COVERAGE:
return None
return mask_bytes
def analyze_colors(
texture_path: Path,
paint_cans: Sequence[PaintCan],
*,
pixel_mask: bytes | None = None,
preserve_chromatic_hue: bool = False,
) -> list[PaintPercentage]:
"""Map non-transparent PNG pixels to their nearest available B42 paint cans."""
if not paint_cans:
raise ValueError("The B42 paint palette is empty")
palette_colors = tuple(
(paint, _srgb_to_lab(paint.rgb), colorsys.rgb_to_hsv(*paint.rgb))
for paint in paint_cans
)
rgba = _read_rgba_png(texture_path)
with rgba:
width, height = rgba.size
if pixel_mask is not None:
if len(pixel_mask) != width * height:
raise ValueError("Variant mask dimensions do not match the texture")
mask = Image.frombytes("L", (width, height), pixel_mask)
rgba.putalpha(ImageChops.multiply(rgba.getchannel("A"), mask))
counts: Counter[PaintCan] = Counter()
rgba_counts = rgba.getcolors(maxcolors=MAX_UNIQUE_COLORS)
if rgba_counts is None:
raise ScanFileError(
f"Image exceeds the {MAX_UNIQUE_COLORS}-unique-color safety limit: {texture_path}"
)
for pixel_count, (red, green, blue, opacity) in rgba_counts:
if not opacity:
continue
paint = _nearest_paint(
red,
green,
blue,
palette_colors,
preserve_chromatic_hue,
)
counts[paint] += pixel_count * opacity
total = sum(counts.values())
if total == 0:
return []
sorted_counts = sorted(counts.items(), key=lambda item: (-item[1], item[0].item_id))
percentage_units = {paint: count * 1000 // total for paint, count in counts.items()}
remaining_units = 1000 - sum(percentage_units.values())
remainders = sorted(
counts,
key=lambda paint: (-(counts[paint] * 1000 % total), paint.item_id),
)
for paint in remainders[:remaining_units]:
percentage_units[paint] += 1
return [
PaintPercentage(paint_can=paint, percentage=percentage_units[paint] / 10)
for paint, _count in sorted_counts
if percentage_units[paint] > 0
]
def scan_workshop(workshop_root: Path, paint_cans: Sequence[PaintCan]) -> ScanReport:
mods = discover_mods(workshop_root)
results: list[TextureResult] = []
warnings: list[str] = []
cars_found = 0
paint_cache: dict[Path, tuple[PaintPercentage, ...]] = {}
decoded_pixels = 0
for mod in mods:
scripts_root = mod.content_root / "media" / "scripts" / "vehicles"
vehicles = parse_vehicle_scripts(scripts_root)
cars_found += len(vehicles)
for vehicle in vehicles:
resolved_skins: list[tuple[SkinDefinition, Path, int]] = []
for skin in vehicle.skins:
reference = skin.texture_reference
texture_path = resolve_texture(mod, reference)
if texture_path is None:
if len(warnings) >= MAX_WARNINGS:
raise ValueError(f"Scan exceeded the {MAX_WARNINGS}-warning safety limit")
warnings.append(f"Missing texture for {vehicle.vehicle_id}: {reference}")
continue
try:
pixel_count = _png_pixel_count(texture_path)
decoded_pixels = _add_decoded_pixels(decoded_pixels, pixel_count)
except (
OSError,
ScanFileError,
UnidentifiedImageError,
Image.DecompressionBombError,
Image.DecompressionBombWarning,
) as error:
if len(warnings) >= MAX_WARNINGS:
raise ValueError(f"Scan exceeded the {MAX_WARNINGS}-warning safety limit")
warnings.append(f"Could not read {texture_path}: {error}")
continue
resolved_skins.append((skin, texture_path, pixel_count))
variant_mask = None
try:
mask_pixels = sum(pixel_count for _skin, _path, pixel_count in resolved_skins)
if (
len(resolved_skins) <= MAX_VARIANT_MASK_SKINS
and mask_pixels <= MAX_VARIANT_MASK_PIXELS
):
decoded_pixels = _add_decoded_pixels(decoded_pixels, mask_pixels)
variant_mask = build_variant_mask(
tuple(texture_path for _skin, texture_path, _pixels in resolved_skins)
)
except (
OSError,
ScanFileError,
UnidentifiedImageError,
Image.DecompressionBombError,
Image.DecompressionBombWarning,
):
pass
vehicle_paint_cache: dict[Path, tuple[PaintPercentage, ...]] = {}
for skin, texture_path, _pixel_count in resolved_skins:
reference = skin.texture_reference
try:
cache = vehicle_paint_cache if variant_mask is not None else paint_cache
paints = cache.get(texture_path)
if paints is None:
paints = tuple(
analyze_colors(
texture_path,
paint_cans,
pixel_mask=variant_mask,
preserve_chromatic_hue=variant_mask is not None,
)
)
cache[texture_path] = paints
except (
OSError,
ScanFileError,
UnidentifiedImageError,
Image.DecompressionBombError,
Image.DecompressionBombWarning,
) as error:
if len(warnings) >= MAX_WARNINGS:
raise ValueError(f"Scan exceeded the {MAX_WARNINGS}-warning safety limit")
warnings.append(f"Could not read {texture_path}: {error}")
continue
results.append(
TextureResult(
workshop_id=mod.workshop_id,
mod_id=mod.mod_id,
mod_name=mod.name,
vehicle_id=vehicle.vehicle_id,
skin_index=skin.skin_index,
texture_reference=reference,
texture_path=texture_path,
paints=paints,
)
)
if len(results) > MAX_TEXTURE_RESULTS:
raise ValueError(
f"Scan exceeded the {MAX_TEXTURE_RESULTS}-texture safety limit"
)
return ScanReport(
mods_scanned=len(mods),
cars_found=cars_found,
textures=tuple(results),
warnings=tuple(warnings),
)
def _format_paints(paints: Iterable[PaintPercentage]) -> str:
formatted = ", ".join(
f"{_safe_console_text(paint.paint_can.display_name)} "
f"[{_safe_console_text(paint.paint_can.item_id)}] {paint.percentage:.1f}%"
for paint in paints
)
return formatted or "no visible pixels"
def _lua_string(value: object) -> str:
text = str(value)
if len(text) > MAX_METADATA_CHARS:
raise ValueError("Lua manifest field exceeds the safety limit")
escaped = (
text
.replace("\\", "\\\\")
.replace('"', '\\"')
.replace("\r", "\\r")
.replace("\n", "\\n")
.replace("\t", "\\t")
)
escaped = re.sub(
r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]",
lambda match: f"\\{ord(match.group(0)):03d}",
escaped,
)
return f'"{escaped}"'
def _render_lua_manifest(
report: ScanReport,
paint_cans: Sequence[PaintCan],
) -> str:
bucket_uses = paint_bucket_uses(paint_cans)
lines = [
"-- Generated by scanner/scan_ki5_workshop.py. Do not edit by hand.",
"PaintMyKI5 = PaintMyKI5 or {}",
"PaintMyKI5.VehiclePaintData = {",
" schemaVersion = 2,",
f" bucketUses = {bucket_uses},",
f" modCount = {report.mods_scanned},",
f" carCount = {report.cars_found},",
f" textureCount = {len(report.textures)},",
" paintCans = {",
]
for paint in sorted(paint_cans, key=lambda entry: entry.item_id.casefold()):
red, green, blue = paint.rgb
lines.extend(
[
" {",
f" item = {_lua_string(paint.item_id)},",
f" name = {_lua_string(paint.display_name)},",
f" rgb = {{ {red:.2f}, {green:.2f}, {blue:.2f} }},",
f" useDelta = {paint.use_delta:.3f},",
" },",
]
)
lines.extend([" },", " vehicles = {"])
sorted_textures = sorted(
report.textures,
key=lambda texture: (
texture.vehicle_id.casefold(),
texture.skin_index,
texture.texture_reference.casefold(),
texture.workshop_id.casefold(),
texture.mod_id.casefold(),
),
)
current_vehicle: str | None = None
for texture in sorted_textures:
if texture.vehicle_id != current_vehicle:
if current_vehicle is not None:
lines.extend([" },", " },"])
current_vehicle = texture.vehicle_id
lines.extend(
[
f" [{_lua_string(texture.vehicle_id)}] = {{",
f" vehicleId = {_lua_string(texture.vehicle_id)},",
f" workshopId = {_lua_string(texture.workshop_id)},",
f" modId = {_lua_string(texture.mod_id)},",
f" modName = {_lua_string(texture.mod_name)},",
" textures = {",
]
)
lines.extend(
[
" {",
f" skinIndex = {texture.skin_index},",
f" texture = {_lua_string(texture.texture_reference)},",
" paints = {",
]
)
for paint, uses in allocate_paint_uses(texture.paints, bucket_uses):
lines.append(
" { item = "
f"{_lua_string(paint.paint_can.item_id)}, percent = {paint.percentage:.1f}, "
f"uses = {uses:.2f} }},"
)
lines.extend([" },", " },"])
if current_vehicle is not None:
lines.extend([" },", " },"])
lines.extend([" },", "}", ""])
manifest = "\n".join(lines)
if len(manifest.encode("utf-8")) > MAX_MANIFEST_BYTES:
raise ValueError(
f"Lua manifest exceeds the {MAX_MANIFEST_BYTES}-byte safety limit"
)
return manifest
def write_lua_manifest(
output_path: Path,
report: ScanReport,
paint_cans: Sequence[PaintCan],
) -> None:
"""Atomically write a deterministic shared-Lua manifest for the in-game menu."""
output_path = output_path.expanduser()
temporary_path: Path | None = None
try:
manifest = _render_lua_manifest(report, paint_cans)
output_path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
newline="\n",
dir=output_path.parent,
prefix=f".{output_path.name}.",
suffix=".tmp",
delete=False,
) as temporary_file:
temporary_path = Path(temporary_file.name)
temporary_file.write(manifest)
temporary_file.flush()
os.fsync(temporary_file.fileno())
os.replace(temporary_path, output_path)
except OSError as error:
raise ValueError(f"Could not write Lua manifest {output_path}: {error}") from error
finally:
try:
if temporary_path is not None:
temporary_path.unlink(missing_ok=True)
except OSError:
pass
def _safe_console_text(value: object) -> str:
text = str(value)
return "".join(
character
if character in "\t" or 0x20 <= ord(character) <= 0x7E or ord(character) >= 0xA0
else "?"
for character in text
)
def _print_report(report: ScanReport, paint_cans: Sequence[PaintCan], output: TextIO) -> None:
print(f"B42 paint cans: {len(paint_cans)}", file=output)
for paint in paint_cans:
red, green, blue = (round(channel * 255) for channel in paint.rgb)
print(
f" {_safe_console_text(paint.display_name)} "
f"[{_safe_console_text(paint.item_id)}] RGB({red}, {green}, {blue})",
file=output,
)
print(file=output)
previous_mod: tuple[str, str] | None = None
previous_vehicle: str | None = None
for texture in report.textures:
mod_key = (texture.workshop_id, texture.mod_id)
if mod_key != previous_mod:
if previous_mod is not None:
print(file=output)
print(
f"Workshop {_safe_console_text(texture.workshop_id)} | "
f"{_safe_console_text(texture.mod_name)} ({_safe_console_text(texture.mod_id)})",
file=output,
)
previous_mod = mod_key
previous_vehicle = None
if texture.vehicle_id != previous_vehicle:
print(f" Vehicle: {_safe_console_text(texture.vehicle_id)}", file=output)
previous_vehicle = texture.vehicle_id
print(f" Texture: {_safe_console_text(texture.texture_reference)}", file=output)
print(f" File: {_safe_console_text(texture.texture_path)}", file=output)
print(f" Nearest B42 paints: {_format_paints(texture.paints)}", file=output)
for warning in report.warnings:
print(f"WARNING: {_safe_console_text(warning)}", file=output)
print(
f"Scanned {report.mods_scanned} mods, {report.cars_found} cars, "
f"{len(report.textures)} textures; {len(report.warnings)} warnings.",
file=output,
)
def build_argument_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="List DamnLib KI5 car skin textures and their color percentages."
)
parser.add_argument(
"--workshop-root",
type=Path,
default=DEFAULT_WORKSHOP_ROOT,
help=f"Project Zomboid Workshop content directory (default: {DEFAULT_WORKSHOP_ROOT})",
)
parser.add_argument(
"--output",
type=Path,
default=DEFAULT_OUTPUT_PATH,
help=f"Generated shared-Lua data file (default: {DEFAULT_OUTPUT_PATH})",
)
parser.add_argument(
"--quiet",
action="store_true",
help="Only print the output path and final scan counts.",
)
parser.add_argument(
"--game-root",
type=Path,
default=DEFAULT_GAME_ROOT,
help=f"Project Zomboid game directory (default: {DEFAULT_GAME_ROOT})",
)
return parser
def main(arguments: Sequence[str] | None = None, *, output: TextIO = sys.stdout) -> int:
options = build_argument_parser().parse_args(arguments)
try:
paint_cans = discover_paint_cans(options.game_root)
report = scan_workshop(options.workshop_root, paint_cans)
write_lua_manifest(options.output, report, paint_cans)
except ValueError as error:
print(f"ERROR: {_safe_console_text(error)}", file=output)
return 2
if not options.quiet:
_print_report(report, paint_cans, output)
print(
f"Wrote {len(report.textures)} textures for {report.cars_found} cars to "
f"{_safe_console_text(options.output.resolve())}",
file=output,
)
return 0
if __name__ == "__main__":
raise SystemExit(main())