Factorio Forums

Code: Select all

Segmented units count doesn't match. Current: {full=294, min=5, asleep=48}. Expected: {full=0, min=5, asleep=48}.

Code: Select all

#!/usr/bin/env python3
"""
Factorio Save Patcher — "Segmented units count doesn't match" fix
=================================================================

Workaround for a Factorio 2.0 segmented-unit save corruption issue (see issue #126642).
counts in the save don't match the actual entity count, causing
a crash on load:

    Error Map.cpp: Segmented units count doesn't match.
    Current:  {full=XXX, min=YYY, asleep=ZZZ}
    Expected: {full=0, min=YYY, asleep=ZZZ}

Usage:
    python fix_segmented_units.py save.zip
    python fix_segmented_units.py save.zip --output fixed.zip
    python fix_segmented_units.py save.zip --scan
    python fix_segmented_units.py save.zip --scan --min 5 --asleep 48
    python fix_segmented_units.py save.zip --info

The script scans level-init.dat and level.dat0 for u32 triplets
(full, min, asleep) and patches any non-zero 'full' values to 0.

Works for ANY triplet values — pass --min and --asleep from your
error message to narrow the search:

    "Current: {full=135, min=15, asleep=43}"
    --> python fix_segmented_units.py save.zip --scan --min 15 --asleep 43

Requirements: Python 3.8+, no external dependencies.
"""

import zipfile
import zlib
import struct
import sys
import os
from pathlib import Path


# ─── Utilities ────────────────────────────────────────────────────────────────

def read_u32(data: bytes, offset: int) -> int:
    return struct.unpack_from('<I', data, offset)[0]


def write_u32(value: int) -> bytes:
    return struct.pack('<I', value)


def hexdump_context(data: bytes, offset: int, radius: int = 32) -> str:
    start = max(0, offset - radius)
    end = min(len(data), offset + radius + 4)
    lines = []
    for i in range(start, end, 16):
        chunk = data[i:i+16]
        hex_part = ' '.join(f'{b:02x}' for b in chunk)
        ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
        marker = ' <<<' if i <= offset < i + 16 else ''
        lines.append(f"  {i:08x}: {hex_part:<48s}  {ascii_part}{marker}")
    return '\n'.join(lines)


# ─── Save header parser ──────────────────────────────────────────────────────

def parse_save_header(data: bytes) -> dict:
    pos = 0

    def u8():
        nonlocal pos; v = data[pos]; pos += 1; return v

    def u16():
        nonlocal pos; v = struct.unpack_from('<H', data, pos)[0]; pos += 2; return v

    def u32():
        nonlocal pos; v = struct.unpack_from('<I', data, pos)[0]; pos += 4; return v

    def u32_optim():
        nonlocal pos
        b = data[pos]; pos += 1
        return b if b != 0xFF else u32()

    def u16_optim():
        nonlocal pos
        b = data[pos]; pos += 1
        return b if b != 0xFF else u16()

    def string():
        nonlocal pos
        length = u32_optim()
        s = data[pos:pos+length].decode('utf-8', errors='replace')
        pos += length
        return s

    header = {}
    header['version'] = (u16(), u16(), u16(), u16())
    pos += 1  # unused bool
    header['campaign'] = string()
    header['name'] = string()
    header['base_mod'] = string()
    header['difficulty'] = u8()
    pos += 2  # finished, player_won
    header['next_level'] = string()
    pos += 3  # can_continue, finished_but_continuing, saving_replay
    pos += 1  # allow_non_admin_debug
    header['loaded_from'] = (u16_optim(), u16_optim(), u16_optim())

    if header['version'][0] >= 2:
        header['build'] = u32()
    else:
        header['build'] = u16()

    header['allowed_commands'] = u8()
    if header['version'][0] >= 2:
        pos += 4  # unknown bytes

    num_mods = u32_optim()
    mods = []
    for _ in range(num_mods):
        mod_name = string()
        mod_ver = (u16_optim(), u16_optim(), u16_optim())
        mod_crc = u32()
        mods.append({'name': mod_name, 'version': mod_ver, 'crc': mod_crc})
    header['mods'] = mods
    header['header_size'] = pos
    return header


# ─── Triplet search ──────────────────────────────────────────────────────────

def find_triplets(data: bytes, expected_min: int = None, expected_asleep: int = None,
                  max_results: int = 100) -> list:
    """
    Find all u32 triplets (full, min, asleep) in binary data.

    If expected_min / expected_asleep are given, only returns triplets
    matching those values for min and asleep. Otherwise returns ALL
    consecutive u32 triplets — you'll need to identify the right one.
    """
    results = []
    for i in range(len(data) - 12):
        a, b, c = struct.unpack_from('<III', data, i)
        if expected_min is not None and expected_asleep is not None:
            if b == expected_min and c == expected_asleep:
                results.append({'offset': i, 'full': a, 'min': b, 'asleep': c})
                if len(results) >= max_results:
                    break
        else:
            # No filter — return triplets where at least one value is
            # small enough to be a plausible count and the values differ
            if a != b and b != c and 0 <= a <= 100000 and 0 <= b <= 100000 and 0 <= c <= 100000:
                results.append({'offset': i, 'full': a, 'min': b, 'asleep': c})
                if len(results) >= max_results:
                    break
    return results


# ─── Segment helpers ─────────────────────────────────────────────────────────

def open_save(save_path: str) -> zipfile.ZipFile:
    if not os.path.exists(save_path):
        print(f"Error: file not found: {save_path}", file=sys.stderr)
        sys.exit(1)
    return zipfile.ZipFile(save_path, 'r')


def find_prefix(zf: zipfile.ZipFile) -> str:
    for name in zf.namelist():
        if name.endswith('/level-init.dat'):
            return name.rsplit('/level-init.dat', 1)[0] + '/'
    return ''


def get_segment_files(zf: zipfile.ZipFile, prefix: str) -> dict:
    segments = {}
    for name in zf.namelist():
        base = name[len(prefix):] if name.startswith(prefix) else name
        if base == 'level-init.dat':
            segments[-1] = name
        elif base.startswith('level.dat') and base[9:].isdigit():
            idx = int(base[9:])
            segments[idx] = name
    return dict(sorted(segments.items()))


def decompress_segment(zf: zipfile.ZipFile, filename: str) -> bytes:
    raw = zf.read(filename)
    if filename.endswith('level-init.dat'):
        return raw
    try:
        return zlib.decompress(raw)
    except zlib.error:
        return raw


# ─── Commands ────────────────────────────────────────────────────────────────

def cmd_scan(save_path: str, expected_min: int, expected_asleep: int):
    """Scan level-init.dat + level.dat0 for segmented unit count triplets."""
    print(f"\n=== Scan: {save_path} ===")

    if expected_min is not None and expected_asleep is not None:
        print(f"  Filter: min={expected_min}, asleep={expected_asleep}")
        print(f"  (values from your error message)\n")
    else:
        print(f"  No --min/--asleep filter. Showing all plausible triplets.\n")

    zf = open_save(save_path)
    prefix = find_prefix(zf)
    segments = get_segment_files(zf, prefix)

    init_data = zf.read(segments[-1])
    header = parse_save_header(init_data)
    ver = '.'.join(str(x) for x in header['version'])
    print(f"  Factorio version: {ver}")
    print(f"  Scenario: {header['name']}")
    print(f"  Mods: {len(header['mods'])}")
    for m in header['mods']:
        v = '.'.join(str(x) for x in m['version'])
        print(f"    {m['name']:<40s} {v}")
    print()

    scan_indices = [-1, 0]
    for idx in scan_indices:
        if idx not in segments:
            continue
        label = 'level-init.dat' if idx == -1 else f'level.dat{idx}'
        data = decompress_segment(zf, segments[idx])
        triplets = find_triplets(data, expected_min, expected_asleep)

        if not triplets:
            print(f"  {label:<20s}  (no matching triplets found)")
            continue

        for t in triplets:
            flag = '  *** SUSPECT (full != 0)' if t['full'] != 0 else '  (full=0, OK)'
            print(f"  {label:<20s} offset 0x{t["offset"]:08x}: "
                  f"full={t["full"]}, min={t["min"]}, asleep={t["asleep"]}{flag}")

    zf.close()
    print()


def cmd_info(save_path: str):
    """Show save file information."""
    print(f"\n=== Save Info: {save_path} ===\n")

    zf = open_save(save_path)
    prefix = find_prefix(zf)
    segments = get_segment_files(zf, prefix)

    init_data = zf.read(segments[-1])
    header = parse_save_header(init_data)

    ver = '.'.join(str(x) for x in header['version'])
    lf = '.'.join(str(x) for x in header['loaded_from'])
    print(f"  Version:           {ver}")
    print(f"  Loaded from:       {lf} (build {header['build']})")
    print(f"  Scenario:          {header['name']}")
    print(f"  Campaign:          {header['campaign'] or '(empty)'}")
    print(f"  Difficulty:        {header['difficulty']}")
    print(f"  Header size:       {header['header_size']} bytes")
    print(f"  Mods:              {len(header['mods'])}")

    for m in header['mods']:
        v = '.'.join(str(x) for x in m['version'])
        print(f"    {m['name']:<40s} {v}")

    dat_segs = {k: v for k, v in segments.items() if k >= 0}
    total_compressed = sum(zf.getinfo(v).file_size for v in dat_segs.values())
    print(f"\n  Segments:          {len(dat_segs)} level.dat files")
    print(f"  Compressed total:  {total_compressed:,} bytes")

    zf.close()
    print()


def cmd_patch(save_path: str, output_path: str, expected_min: int, expected_asleep: int):
    """Patch segmented unit counts in level.dat0."""
    print(f"\n=== Patch: {save_path} ===\n")

    zf = open_save(save_path)
    prefix = find_prefix(zf)
    segments = get_segment_files(zf, prefix)

    init_data = zf.read(segments[-1])
    header = parse_save_header(init_data)
    ver = '.'.join(str(x) for x in header['version'])
    print(f"  Factorio version: {ver}")
    print(f"  Mods: {len(header['mods'])}")

    dat0_name = segments.get(0)
    if not dat0_name:
        print("  Error: level.dat0 not found!")
        zf.close()
        return

    raw_dat0 = zf.read(dat0_name)
    decomp_dat0 = bytearray(zlib.decompress(raw_dat0))

    triplets = find_triplets(decomp_dat0, expected_min, expected_asleep)
    suspect = [t for t in triplets if t['full'] != 0]

    if not suspect:
        print("\n  No non-zero 'full' triplets found in level.dat0.")
        if expected_min is None or expected_asleep is None:
            print("  Try: --scan --min <VALUE> --asleep <VALUE> from your error message.")
        else:
            print("  Try --scan to see all triplets, or check other level.dat segments.")
        zf.close()
        return

    print(f"\n  Found {len(suspect)} suspect triplet(s) in level.dat0:\n")

    for i, t in enumerate(suspect):
        print(f"  [{i+1}] offset 0x{t['offset']:08x}: "
              f"full={t['full']}, min={t['min']}, asleep={t['asleep']}")
        print(hexdump_context(decomp_dat0, t['offset'], 32))
        print()

    if len(suspect) > 1:
        print(f"  WARNING: {len(suspect)} non-zero triplets found. Patching ALL.")

    for t in suspect:
        off = t['offset']
        old_val = read_u32(decomp_dat0, off)
        print(f"  Patching offset 0x{off:08x}: {old_val} -> 0")
        decomp_dat0[off:off+4] = write_u32(0)
        print(hexdump_context(decomp_dat0, off, 32))
        print()

    # Repack
    recomp = zlib.compress(bytes(decomp_dat0))
    print(f"  Compressed: {len(raw_dat0):,} -> {len(recomp):,} bytes")

    # Verify
    verify = zlib.decompress(recomp)
    for t in suspect:
        val = read_u32(verify, t['offset'])
        assert val == 0, f"Verification failed at 0x{t['offset']:x}!"
    print("  Verification passed")

    # Write
    print(f"\n  Writing: {output_path}")
    with zipfile.ZipFile(output_path, 'w', zipfile.ZIP_DEFLATED) as zf_out:
        for item in zf.infolist():
            if item.filename == dat0_name:
                zf_out.writestr(item, recomp)
                print(f"    PATCHED: {item.filename}")
            else:
                data = zf.read(item.filename)
                zf_out.writestr(item, data)

    zf.close()
    print(f"\n=== DONE ===")
    print(f"  Patched save: {output_path}")
    print(f"  Load it in Factorio and check if the crash is gone.\n")


# ─── CLI ─────────────────────────────────────────────────────────────────────

def main():
    if len(sys.argv) < 2 or '--help' in sys.argv or '-h' in sys.argv:
        print(__doc__)
        sys.exit(0)

    save_path = sys.argv[1]

    # Parse --min and --asleep from args
    expected_min = None
    expected_asleep = None
    for i, arg in enumerate(sys.argv):
        if arg == '--min' and i + 1 < len(sys.argv):
            expected_min = int(sys.argv[i + 1])
        if arg == '--asleep' and i + 1 < len(sys.argv):
            expected_asleep = int(sys.argv[i + 1])

    if '--scan' in sys.argv:
        cmd_scan(save_path, expected_min, expected_asleep)
    elif '--info' in sys.argv:
        cmd_info(save_path)
    else:
        output = None
        for i, arg in enumerate(sys.argv):
            if arg == '--output' and i + 1 < len(sys.argv):
                output = sys.argv[i + 1]
        if not output:
            stem = Path(save_path).stem
            suffix = Path(save_path).suffix
            output = str(Path(save_path).parent / f"{stem} patched{suffix}")
        cmd_patch(save_path, output, expected_min, expected_asleep)


if __name__ == '__main__':
    main()

Code: Select all

python fix_segmented_units.py "YOUR SAVE NAME.zip" --scan --min YOUR_MIN_NUM --asleep YOUR_ASLEEP_NUM

Code: Select all

python fix_segmented_units.py "YOUR SAVE NAME.zip" --min YOUR_MIN_NUM --asleep YOUR_ASLEEP_NUM