fh3_azimuth_converter.py
Nov 2, 2026, 07:21 AM 20.3 KB
#!/usr/bin/env python3
"""
Forza Horizon 3 (UWP / Xbox app) -> Azimuth Save Converter
A small desktop app: pick your UWP save folder and your Azimuth saves folder, choose what to
convert, press Convert.
SAFETY
* The UWP folder is only ever READ.
* Azimuth's saves are only ADDED to (a new revision per container). Nothing is edited or deleted.
* Your existing Azimuth saves are zipped to Documents\\SaveTransfer before anything is written.
* Every run writes an undo list; the "Undo a previous conversion" button uses it.
Run: python fh3_azimuth_converter.py (needs Python 3.8+ with tkinter, no other packages)
Build: pyinstaller --onefile --windowed fh3_azimuth_converter.py
"""
import hashlib
import os
import queue
import re
import shutil
import struct
import subprocess
import sys
import threading
import time
import uuid
import zipfile
from pathlib import Path
APP_TITLE = "FH3 UWP -> Azimuth Save Converter"
FH3_PKG = "Microsoft.OpusPG_8wekyb3d8bbwe"
ACCT_RE = re.compile(r"^[0-9A-Fa-f]{16}_[0-9A-Fa-f]{32}$")
REV_RE = re.compile(r"^[0-9a-f]{16}$")
BACKUP_DIR = Path.home() / "Documents" / "SaveTransfer"
DEFAULT_PLAYER = "player-1"
DEFAULT_USER = "User_9000000000000"
# (key, label, container-name prefixes, selected by default)
CATEGORIES = [
("profile", "Profile and progress (replaces your current Azimuth profile)", ("User_",), True),
("liveries", "Livery designs", ("Livery_",), True),
("base", "Base and soul-bound liveries", ("BaseLivery_", "SoulBoundLivery_"), True),
("tunes", "Tunes", ("Tuning_",), True),
("thumbs", "Garage thumbnails", ("GarageThumbnail_",), True),
]
class ConvertError(Exception):
pass
# ============================================================ UWP side (read-only)
class Entry:
def __init__(self, name, num, folder, size):
self.name, self.num, self.folder, self.size = name, num, folder, size
def default_uwp_path():
base = os.environ.get("LOCALAPPDATA", "")
return str(Path(base) / "Packages" / FH3_PKG / "SystemAppData" / "wgs") if base else ""
def resolve_uwp(path):
"""Accept the wgs folder, the account folder inside it, or the whole package folder."""
p = Path(path)
for cand in (p, p / "wgs", p / "SystemAppData" / "wgs"):
if ACCT_RE.match(cand.name) and (cand / "containers.index").is_file():
return cand
if cand.is_dir():
hits = [d for d in cand.iterdir()
if d.is_dir() and ACCT_RE.match(d.name) and (d / "containers.index").is_file()]
if len(hits) == 1:
return hits[0]
if len(hits) > 1:
raise ConvertError("More than one account folder found. Select the one you want:\n"
+ "\n".join(h.name for h in hits))
raise ConvertError("No FH3 save found there (no containers.index).\n"
"Select the 'wgs' folder under\n"
"%LOCALAPPDATA%\\Packages\\Microsoft.OpusPG_8wekyb3d8bbwe\\SystemAppData\\wgs")
def read_index(path):
d = Path(path).read_bytes()
pos = 0
def need(n):
if pos + n > len(d):
raise ConvertError("containers.index is truncated or in an unexpected format.")
def u32():
nonlocal pos
need(4)
v = struct.unpack_from("<I", d, pos)[0]
pos += 4
return v
def skip(n):
nonlocal pos
need(n)
pos += n
def wstr():
nonlocal pos
n = u32()
if n > 4096:
raise ConvertError("containers.index has an implausible string length.")
need(2 * n)
s = d[pos:pos + 2 * n].decode("utf-16le")
pos += 2 * n
return s
version, count, _flags = u32(), u32(), u32()
if version != 14:
raise ConvertError(f"containers.index version {version} (expected 14) is not supported.")
wstr()
skip(8)
u32()
wstr()
skip(8)
entries = []
for _ in range(count):
name, _n2, _etag = wstr(), wstr(), wstr()
need(1)
num = d[pos]
pos += 1
u32()
need(16)
folder = uuid.UUID(bytes_le=d[pos:pos + 16]).hex.upper()
pos += 16
skip(16)
need(8)
size = struct.unpack_from("<Q", d, pos)[0]
pos += 8
entries.append(Entry(name, num, folder, size))
return entries
def read_container(folder, num):
cf = folder / f"container.{num}"
if not cf.is_file():
return None
data = cf.read_bytes()
if len(data) < 8:
raise ConvertError(f"{cf} is too small.")
ver, count = struct.unpack_from("<II", data, 0)
if ver != 4 or len(data) != 8 + 160 * count:
raise ConvertError(f"{cf}: unexpected container format.")
out = []
for i in range(count):
off = 8 + 160 * i
name = data[off:off + 128].decode("utf-16le").split("\x00")[0]
g_cloud = uuid.UUID(bytes_le=data[off + 128:off + 144]).hex.upper()
g_local = uuid.UUID(bytes_le=data[off + 144:off + 160]).hex.upper()
path = folder / g_local
if not path.is_file():
path = folder / g_cloud
out.append((name, path))
return out
# ============================================================ Azimuth side (additive only)
def resolve_azimuth_saves(path):
"""Accept the Azimuth install folder, a title folder, the saves folder, or player folder."""
p = Path(path)
if p.name.startswith("player-") and p.parent.name == "saves":
return p.parent
if p.name == "saves" and p.is_dir():
return p
if (p / "saves").is_dir():
return p / "saves"
titles = p / "data" / "titles"
if titles.is_dir():
found = [t / "saves" for t in titles.iterdir() if (t / "saves").is_dir()]
if len(found) == 1:
return found[0]
if len(found) > 1:
preferred = [f for f in found if f.parent.name.upper() == "4CE1DFAB"]
if len(preferred) == 1:
return preferred[0]
raise ConvertError("Several game folders found; select the 'saves' folder of the FH3 one "
"(the folder named 4CE1DFAB).")
raise ConvertError("Couldn't find Azimuth's saves folder there.\n"
"Start FH3 in Azimuth once so it creates it, then select\n"
"...\\azimuth-windows\\data\\titles\\4CE1DFAB\\saves")
def detect_player(saves):
players = [d.name for d in saves.iterdir() if d.is_dir() and d.name.startswith("player-")] \
if saves.is_dir() else []
return players[0] if len(players) == 1 else DEFAULT_PLAYER
def detect_user_container(saves, player):
base = saves / player
users = [d.name for d in base.iterdir() if d.is_dir() and d.name.startswith("User_")] \
if base.is_dir() else []
return users[0] if len(users) == 1 else DEFAULT_USER
def make_commit(blobs):
text = "Azimuth complete save revision v2\n" + f"blobs {len(blobs)}\n"
text += "".join(f"{sha} {size} {name}\n" for sha, size, name in blobs)
return text.encode("utf-8")
def next_revision(cdir):
rd = cdir / "revisions"
nums = [int(d.name, 16) for d in rd.iterdir() if d.is_dir() and REV_RE.match(d.name)] \
if rd.is_dir() else []
return (max(nums) + 1) if nums else 1
def backup_dir(saves):
if not saves.is_dir() or not any(saves.rglob("*")):
return None
BACKUP_DIR.mkdir(parents=True, exist_ok=True)
out = BACKUP_DIR / f"azimuth_saves_pre-convert_{time.strftime('%Y%m%d-%H%M%S')}.zip"
with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as z:
for f in saves.rglob("*"):
if f.is_file():
z.write(f, f.relative_to(saves.parent).as_posix())
return out
def prune(rev):
"""Remove empty revisions/ and container folders left behind after deleting `rev`."""
for d in (rev.parent, rev.parent.parent):
try:
d.rmdir()
except OSError:
break
# ============================================================ Planning and running
class PlanItem:
def __init__(self, source_name, target, blobs, cdir):
self.source_name, self.target, self.blobs, self.cdir = source_name, target, blobs, cdir
def build_plan(acct, saves, prefixes, skip_existing, log):
entries = read_index(acct / "containers.index")
player = detect_player(saves)
user_target = detect_user_container(saves, player)
plan, missing, existing = [], 0, 0
for e in entries:
if not e.name.startswith(prefixes):
continue
is_profile = e.name.startswith("User_")
folder = acct / e.folder
blobs = read_container(folder, e.num) if folder.is_dir() else None
if blobs is None or any(not p.is_file() for _, p in blobs):
missing += 1
if missing <= 5:
log(f" skipped {e.name}: files missing on the UWP side")
continue
target = user_target if is_profile else e.name
cdir = saves / player / target
if skip_existing and not is_profile and (cdir / "revisions").is_dir():
existing += 1
continue
plan.append(PlanItem(e.name, target, blobs, cdir))
if missing > 5:
log(f" ... {missing - 5} more skipped because of missing files")
if existing:
log(f" {existing} already exist in Azimuth and were left untouched")
return plan
def summarize(plan):
counts = {}
for it in plan:
key = it.source_name.split("_")[0]
counts[key] = counts.get(key, 0) + 1
return ", ".join(f"{n} {k}" for k, n in sorted(counts.items())) or "nothing"
def run_conversion(plan, saves, log, progress, cancel):
bk = backup_dir(saves)
log(f"Backed up current Azimuth saves -> {bk}" if bk else "No existing Azimuth saves to back up.")
written = []
try:
for i, it in enumerate(plan, 1):
if cancel.is_set():
raise ConvertError("cancelled")
rev = it.cdir / "revisions" / f"{next_revision(it.cdir):016x}"
(rev / "data").mkdir(parents=True)
written.append(rev)
meta = []
for name, path in it.blobs:
data = path.read_bytes()
(rev / "data" / name).write_bytes(data)
meta.append((hashlib.sha256(data).hexdigest(), len(data), name))
(rev / "commit").write_bytes(make_commit(meta)) # commit last = revision complete
progress(i, len(plan))
except Exception as ex:
for r in reversed(written):
shutil.rmtree(r, ignore_errors=True)
prune(r)
raise ConvertError(f"Stopped ({ex}). Everything written in this run was rolled back.") from ex
BACKUP_DIR.mkdir(parents=True, exist_ok=True)
undo = BACKUP_DIR / f"azimuth_convert_undo_{time.strftime('%Y%m%d-%H%M%S')}.txt"
undo.write_text("\n".join(str(r) for r in written) + "\n", encoding="utf-8")
log(f"Wrote {len(written)} revisions. Undo list: {undo}")
return undo
def undo_from_file(undo_path, log):
removed = 0
for line in Path(undo_path).read_text(encoding="utf-8").splitlines():
rev = Path(line.strip())
if (line.strip() and rev.parent.name == "revisions" and REV_RE.match(rev.name)
and (rev / "commit").is_file()):
shutil.rmtree(rev, ignore_errors=True)
prune(rev)
removed += 1
elif line.strip():
log(f" skipped (not found or not a revision): {line.strip()}")
log(f"Undo complete: removed {removed} revision(s).")
return removed
def running_games():
if os.name != "nt":
return []
try:
flags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
out = subprocess.run(["tasklist"], capture_output=True, text=True,
creationflags=flags).stdout.lower()
except Exception:
return []
return [w for w in ("forza", "azimuth") if w in out]
# ============================================================ GUI
def main_gui():
import tkinter as tk
from tkinter import filedialog, messagebox, scrolledtext, ttk
root = tk.Tk()
root.title(APP_TITLE)
root.geometry("860x700")
root.minsize(760, 600)
pad = {"padx": 10, "pady": 4}
uwp_var = tk.StringVar(value=default_uwp_path())
az_var = tk.StringVar()
skip_var = tk.BooleanVar(value=True)
cat_vars = {k: tk.BooleanVar(value=d) for k, _, _, d in CATEGORIES}
msgs = queue.Queue()
cancel = threading.Event()
busy = {"on": False}
ttk.Label(root, text="Move your Forza Horizon 3 (Xbox app / UWP) save into Azimuth",
font=("Segoe UI", 13, "bold")).pack(anchor="w", **pad)
ttk.Label(root, text="Close FH3 and Azimuth first. Your UWP save is never modified.",
foreground="#555").pack(anchor="w", padx=10)
def path_row(parent, title, var, browse_title, auto=None):
box = ttk.LabelFrame(parent, text=title)
box.pack(fill="x", **pad)
row = ttk.Frame(box)
row.pack(fill="x", padx=8, pady=6)
ttk.Entry(row, textvariable=var).pack(side="left", fill="x", expand=True)
def browse():
d = filedialog.askdirectory(title=browse_title, initialdir=var.get() or None)
if d:
var.set(d)
ttk.Button(row, text="Browse...", command=browse).pack(side="left", padx=(8, 0))
if auto:
ttk.Button(row, text="Auto-detect", command=auto).pack(side="left", padx=(6, 0))
path_row(root, "1. UWP save folder (the 'wgs' folder)", uwp_var, "Select the UWP 'wgs' folder",
auto=lambda: uwp_var.set(default_uwp_path()))
path_row(root, "2. Azimuth folder (install folder or its 'saves' folder)", az_var,
"Select the Azimuth folder")
opts = ttk.LabelFrame(root, text="3. What to convert")
opts.pack(fill="x", **pad)
for key, label, _, _ in CATEGORIES:
ttk.Checkbutton(opts, text=label, variable=cat_vars[key]).pack(anchor="w", padx=8, pady=1)
ttk.Checkbutton(opts, text="Keep anything Azimuth already has for liveries, tunes and thumbnails "
"(recommended)", variable=skip_var).pack(anchor="w", padx=8, pady=(6, 4))
bar = ttk.Frame(root)
bar.pack(fill="x", **pad)
prog = ttk.Progressbar(root, mode="determinate")
prog.pack(fill="x", padx=10, pady=(0, 4))
log_box = scrolledtext.ScrolledText(root, height=12, state="disabled", font=("Consolas", 9))
log_box.pack(fill="both", expand=True, padx=10, pady=(0, 10))
def log(text):
msgs.put(("log", text))
def append(text):
log_box.configure(state="normal")
log_box.insert("end", text + "\n")
log_box.see("end")
log_box.configure(state="disabled")
buttons = []
def set_busy(on):
busy["on"] = on
for b in buttons:
b.configure(state="disabled" if on else "normal")
def poll():
try:
while True:
kind, *rest = msgs.get_nowait()
if kind == "log":
append(rest[0])
elif kind == "progress":
prog.configure(maximum=rest[1], value=rest[0])
elif kind == "done":
set_busy(False)
if rest[0]:
messagebox.showinfo(APP_TITLE, rest[0])
elif kind == "error":
set_busy(False)
append("ERROR: " + rest[0])
messagebox.showerror(APP_TITLE, rest[0])
except queue.Empty:
pass
root.after(100, poll)
def gather():
prefixes = tuple(p for k, _, ps, _ in CATEGORIES if cat_vars[k].get() for p in ps)
if not prefixes:
raise ConvertError("Select at least one thing to convert.")
acct = resolve_uwp(uwp_var.get().strip().strip('"'))
if not az_var.get().strip():
raise ConvertError("Select your Azimuth folder.")
saves = resolve_azimuth_saves(az_var.get().strip().strip('"'))
return acct, saves, prefixes
def start(dry_run):
if busy["on"]:
return
try:
acct, saves, prefixes = gather()
except ConvertError as ex:
messagebox.showerror(APP_TITLE, str(ex))
return
if not dry_run:
warn = running_games()
if warn and not messagebox.askyesno(
APP_TITLE, "These look like they are still running: " + ", ".join(warn) +
".\nClose them first, or the game may overwrite the converted save.\n\nContinue anyway?",
default="no"):
return
text = ("This adds the selected saves to Azimuth as new revisions.\n"
"Your current Azimuth saves are backed up first, and you can undo afterwards.")
if cat_vars["profile"].get():
text += ("\n\nThe PROFILE will replace your current Azimuth progress "
"(the old one stays as an older revision).")
if not messagebox.askokcancel(APP_TITLE, text + "\n\nContinue?"):
return
cancel.clear()
prog.configure(value=0)
set_busy(True)
def work():
try:
log(f"UWP account folder : {acct.name}")
log(f"Azimuth saves : {saves}")
plan = build_plan(acct, saves, prefixes, skip_var.get(), log)
if not plan:
msgs.put(("done", "Nothing to convert with these settings."))
return
log(f"{'Preview: would convert' if dry_run else 'Converting'} {len(plan)} "
f"container(s): {summarize(plan)}")
if dry_run:
msgs.put(("done", f"Preview only - nothing was written.\n\nWould convert {len(plan)} "
f"container(s):\n{summarize(plan)}"))
return
undo = run_conversion(plan, saves, log, lambda i, n: msgs.put(("progress", i, n)), cancel)
msgs.put(("done", f"Done. Converted {len(plan)} container(s).\n\nStart FH3 in Azimuth and "
f"check your progress.\n\nTo undo, use 'Undo a previous conversion' "
f"with:\n{undo}"))
except ConvertError as ex:
msgs.put(("error", str(ex)))
except Exception as ex: # unexpected: show, don't crash silently
msgs.put(("error", f"Unexpected error: {ex}"))
threading.Thread(target=work, daemon=True).start()
def undo():
if busy["on"]:
return
f = filedialog.askopenfilename(title="Select an undo list", initialdir=str(BACKUP_DIR),
filetypes=[("Undo lists", "azimuth_convert_undo_*.txt"),
("Text files", "*.txt")])
if not f:
return
if not messagebox.askyesno(APP_TITLE, "Remove every revision this conversion added to "
"Azimuth?\n\nAzimuth goes back to what it had before."):
return
try:
n = undo_from_file(f, lambda t: append(t))
messagebox.showinfo(APP_TITLE, f"Removed {n} revision(s).")
except Exception as ex:
messagebox.showerror(APP_TITLE, f"Undo failed: {ex}")
b1 = ttk.Button(bar, text="Preview (writes nothing)", command=lambda: start(True))
b2 = ttk.Button(bar, text="Convert", command=lambda: start(False))
b3 = ttk.Button(bar, text="Undo a previous conversion...", command=undo)
for b in (b1, b2, b3):
b.pack(side="left", padx=(0, 8))
buttons.append(b)
ttk.Button(bar, text="Open backups folder",
command=lambda: (BACKUP_DIR.mkdir(parents=True, exist_ok=True), os.startfile(BACKUP_DIR))
if os.name == "nt" else None).pack(side="right")
append("Ready. Pick your folders, then press Preview or Convert.")
poll()
root.mainloop()
if __name__ == "__main__":
main_gui()