"""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())