Fix good-first-issue batch (#5-#12)

common/jsonio.py

@@ -4,6 +4,8 @@
 import json
 import math
 import os
+import shutil
+import tempfile
 from pathlib import Path
 from typing import Any
 
@@ -33,9 +35,19 @@ def write_json(path: str | Path, value: Any, indent: int | None = 1, backup: boo
     """Write via a temporary file and atomic replace; optionally keep the previous file as .bak."""
     path = Path(path)
     path.parent.mkdir(parents=True, exist_ok=True)
-    temp = path.with_name(path.name + ".tmp")
-    temp.write_text(dumps(value, indent), encoding="utf-8")
-    if backup and path.exists():
-        os.replace(path, path.with_name(path.name + ".bak"))
-    os.replace(temp, path)
+    text = dumps(value, indent)
+    handle = tempfile.NamedTemporaryFile("w", encoding="utf-8", newline="", dir=path.parent,
+                                         prefix=path.name + ".", suffix=".tmp", delete=False)
+    temp = Path(handle.name)
+    try:
+        with handle:
+            handle.write(text)
+            handle.flush()
+            os.fsync(handle.fileno())
+        if backup and path.exists():
+            shutil.copy2(path, path.with_name(path.name + ".bak"))   # the target itself is never moved away
+        os.replace(temp, path)
+    except BaseException:
+        temp.unlink(missing_ok=True)
+        raise
     return path

common/pipeline/versions.py

@@ -129,7 +129,10 @@ def restore(path: str | Path, version: int) -> dict:
     source = version_file(path, version)
     if not source.exists():
         if source_entry is history["versions"][-1] and path.exists():
-            return source_entry     # already current
+            if _sha256(path) == source_entry.get("sha256"):
+                return source_entry     # already current, verified by hash
+            raise FileNotFoundError(f"{path.name} has changed since version {version} was recorded and that version "
+                                    f"was never archived ({source.name} missing), so it cannot be restored.")
         raise FileNotFoundError(f"Version {version} was never archived ({source.name} missing).")
     archive_current(path)
     shutil.copy2(source, path)

games/ultima-online/outfit-lab/uo_vd_writer.py

@@ -94,7 +94,10 @@ def encode_frame(idx, cx, cy):
             x0 = x
             while x < w and row[x] >= 0 and x - x0 < 0xFFF:
                 x += 1
-            out += struct.pack("<I", (((x0 - cx) & 0x3FF) << 22) | (((y - cy - h) & 0x3FF) << 12) | (x - x0))
+            dx, dy = x0 - cx, y - cy - h
+            if not (-512 <= dx <= 511 and -512 <= dy <= 511):
+                raise ValueError(f"run offset out of 10-bit range (-512..511): dx={dx}, dy={dy} at row {y}")
+            out += struct.pack("<I", ((dx & 0x3FF) << 22) | ((dy & 0x3FF) << 12) | (x - x0))
             out += bytes(row[x0:x].astype(np.uint8))
     out += struct.pack("<I", END)
     return bytes(out)

games/ultima-online/outfit-lab/vd.py

@@ -39,8 +39,10 @@ def read_vd(path):
                 hdr^=(0x200<<22)|(0x200<<12)
                 x=xb+((hdr>>22)&0x3ff); y=yb+((hdr>>12)&0x3ff); n=hdr&0xfff
                 idx=d[p:p+n]; p+=n
-                for k,ci in enumerate(idx):
-                    img[y,x+k,:3]=pal[ci]; img[y,x+k,3]=255
+                if 0<=y<h:                       # clip runs outside the declared size
+                    x0,x1=max(x,0),min(x+n,w)
+                    for k in range(x0-x,x1-x):
+                        img[y,x+k,:3]=pal[idx[k]]; img[y,x+k,3]=255
             frames.append(dict(cx=cx,cy=cy,w=w,h=h,img=img))
         anims[(a,dr)]=frames
     return at,nact,anims

games/ultima-online/region-masks/build_regions.py

@@ -16,6 +16,7 @@
 import numpy as np
 from scipy import ndimage
 from PIL import Image, ImageDraw, ImageFont
+from region_fonts import load_font
 
 SCRIPT_DIR = Path(__file__).resolve().parent
 REPO = SCRIPT_DIR.parents[2]
@@ -106,8 +107,8 @@ def frame_regions(body, garments):
 def build(actions):
     (ROOT / 'frames').mkdir(parents=True, exist_ok=True)
     reader = UOReader(client_source(CLIENT))
-    font = ImageFont.truetype('C:/Windows/Fonts/arial.ttf', 18)
-    small = ImageFont.truetype('C:/Windows/Fonts/arial.ttf', 13)
+    font = load_font(18)
+    small = load_font(13)
     old = ROOT / 'region-report.json'
     report = json.loads(old.read_text()) if old.exists() else {}
     report.update({'regions': REGIONS, 'garments': {k: a for k, a, _ in GARMENTS}})

games/ultima-online/region-masks/clothing_fit.py

@@ -25,6 +25,7 @@
 import numpy as np
 from scipy import ndimage
 from PIL import Image, ImageDraw, ImageFont
+from region_fonts import load_font
 
 SCRIPT_DIR = Path(__file__).resolve().parent
 REPO = SCRIPT_DIR.parents[2]
@@ -251,8 +252,8 @@ def overlay(body_rgba, g, O, L=None, c=None):
 
 
 def sheets(out, label, results, worst, ctx, cand, distinct, actions):
-    font = ImageFont.truetype('C:/Windows/Fonts/arial.ttf', 12)
-    big = ImageFont.truetype('C:/Windows/Fonts/arial.ttf', 16)
+    font = load_font(12)
+    big = load_font(16)
     order = sorted((a for a in distinct if a in results), key=lambda a: -results[a]['deepFailRate'])
     cell = 120; cols = 6; W = cols * (2 * cell + 20) + 20
     H = 70 + ((len(order) + cols - 1) // cols) * (cell + 44)

games/ultima-online/region-masks/package_masks.py

@@ -10,6 +10,7 @@
 import numpy as np
 from scipy import ndimage
 from PIL import Image, ImageDraw, ImageFont
+from region_fonts import load_font
 
 ROOT = Path(__file__).resolve().parent
 sys.path.insert(0, str(ROOT))
@@ -278,8 +279,8 @@ def main():
     for sub in ['sheets', 'gizmos', 'compare']:
         (OUT / sub).mkdir(parents=True)
     names = action_names()
-    font = ImageFont.truetype('C:/Windows/Fonts/arial.ttf', 18)
-    small = ImageFont.truetype('C:/Windows/Fonts/arial.ttf', 12)
+    font = load_font(18)
+    small = load_font(12)
     report = json.loads((ROOT / 'region-report.json').read_text())
     actions = []
     for a in range(35):

games/ultima-online/region-masks/pose_editor_server.py

@@ -1,6 +1,6 @@
 """Loopback-only live pose editor, local saves, and asynchronous Blender export."""
 from pathlib import Path
-import argparse, functools, json, math, os, subprocess, threading, time, uuid
+import argparse, functools, json, math, os, re, shutil, subprocess, threading, time, uuid
 from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
 from urllib.parse import urlparse, parse_qs
 
@@ -9,6 +9,27 @@
 SOURCE=Path(__file__).parent
 EDITABLE={f'{bone}_{side}' for bone in ('upperarm','lowerarm','hand','thigh','calf','foot') for side in ('l','r')}|{'LegPlate_L','LegPlate_R','Hip_L','Hip_R'}
 
+def _version_key(path):
+    return tuple(int(n) for n in re.findall(r'\d+',path.parent.name))
+
+def find_blender(root=None,environ=None):
+    """SPRITEMOTION_BLENDER, then the newest tools/blender-runtime build, then PATH, then Program Files."""
+    root=Path(root) if root else ROOT;environ=os.environ if environ is None else environ
+    exe=environ.get('SPRITEMOTION_BLENDER')
+    if not exe:
+        runtime=sorted((root/'tools/blender-runtime').glob('*/blender.exe'),key=_version_key)
+        exe=str(runtime[-1]) if runtime else None
+    exe=exe or shutil.which('blender')
+    if not exe:
+        found=sorted(Path(environ.get('ProgramFiles','C:/Program Files'),'Blender Foundation').glob('Blender */blender.exe'),key=_version_key)
+        exe=str(found[-1]) if found else None
+    if not exe:raise RuntimeError('Set SPRITEMOTION_BLENDER to blender.exe.')
+    return exe
+
+def write_private(path,text):
+    fd=os.open(path,os.O_WRONLY|os.O_CREAT|os.O_TRUNC,0o600)
+    with os.fdopen(fd,'w') as f:f.write(text)
+
 def validate(doc,scene):
     if not isinstance(doc,dict) or doc.get('version')!=1 or doc.get('assetId')!=scene['assetId']:
         raise ValueError('Edits do not match this model version.')
@@ -61,20 +82,20 @@ def do_POST(self):
             doc=validate(json.loads(self.rfile.read(size)),self.server.scene)
             if self.path=='/api/save':
                 with self.server.lock:
-                    tmp=OUT/'editor/edits.tmp';tmp.write_text(json.dumps(doc,indent=2));os.replace(tmp,OUT/'editor/edits.json')
+                    tmp=OUT/'editor/edits.tmp';write_private(tmp,json.dumps(doc,indent=2));os.replace(tmp,OUT/'editor/edits.json')
                 return self.send_json({'saved':True})
             with self.server.lock:
                 if any(j['status']=='running' for j in self.server.jobs.values()):return self.send_json({'error':'An export is already running.'},409)
                 job=uuid.uuid4().hex[:12];folder=OUT/'editor/exports'/job;folder.mkdir(parents=True)
-                edits=folder/'edits.json';edits.write_text(json.dumps(doc))
+                edits=folder/'edits.json';write_private(edits,json.dumps(doc))
                 self.server.jobs[job]={'id':job,'status':'running'}
             threading.Thread(target=self.bake,args=(job,folder,edits),daemon=True).start()
             self.send_json({'id':job},202)
         except (ValueError,TypeError,KeyError) as e:self.send_json({'error':str(e)},400)
     def bake(self,job,folder,edits):
         output=folder/'UO_Female_Edited.blend'
         try:
-            command=[str(ROOT/'tools/blender-runtime/blender-5.2.2-windows-x64/blender.exe'),'-b',str(OUT/self.server.scene.get('sourceBlend','UO_Female_Idle_Walk_Run.blend')),'--python-exit-code','1','--python',str(ROOT/'tools/blender/apply_pose_edits.py'),'--','--edits',str(edits),'--output',str(output)]
+            command=[find_blender(),'-b',str(OUT/self.server.scene.get('sourceBlend','UO_Female_Idle_Walk_Run.blend')),'--python-exit-code','1','--python',str(ROOT/'tools/blender/apply_pose_edits.py'),'--','--edits',str(edits),'--output',str(output)]
             with (folder/'export.log').open('w') as log:
                 result=subprocess.run(command,stdout=log,stderr=subprocess.STDOUT,timeout=180,creationflags=getattr(subprocess,'CREATE_NO_WINDOW',0))
             if result.returncode or not output.exists():raise RuntimeError('Blender export failed; see the export log.')

games/ultima-online/region-masks/region_fonts.py

@@ -0,0 +1,21 @@
+"""Font loading that works without Arial (Linux, macOS, stripped Windows)."""
+import os
+from pathlib import Path
+from PIL import ImageFont
+
+def candidates():
+    windir = os.environ.get('WINDIR') or os.environ.get('SystemRoot') or 'C:/Windows'
+    return [Path(windir, 'Fonts', 'arial.ttf'), Path('/Library/Fonts/Arial.ttf'),
+            Path('/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf'), Path('/usr/share/fonts/TTF/DejaVuSans.ttf')]
+
+def load_font(size, paths=None):
+    """Return a TrueType font of `size`, or PIL's built-in default when none is installed."""
+    for path in (candidates() if paths is None else paths):
+        try:
+            return ImageFont.truetype(str(path), size)
+        except OSError:
+            continue
+    try:
+        return ImageFont.load_default(size)
+    except TypeError:  # Pillow < 10.1 has no sized default
+        return ImageFont.load_default()

tests/test_pose_editor.py

@@ -1,6 +1,9 @@
 """Edits accepted for Blender export must be bounded and tied to the exact asset."""
 import importlib.util
+import os
+import tempfile
 import unittest
+from unittest import mock
 from pathlib import Path
 
 path=Path(__file__).resolve().parents[1]/'games/ultima-online/region-masks/pose_editor_server.py'
@@ -28,4 +31,37 @@ def test_nonfinite_and_nonunit_quaternions_rejected(self):
                 self.doc['edits']['Walk:9']['upperarm_l']=q
                 with self.assertRaises(ValueError):module.validate(self.doc,self.scene)
 
+class FindBlender(unittest.TestCase):
+    def make(self,root,*names):
+        for name in names:
+            exe=Path(root)/'tools/blender-runtime'/name/'blender.exe';exe.parent.mkdir(parents=True);exe.touch()
+    def test_env_wins(self):
+        with tempfile.TemporaryDirectory() as d:
+            self.make(d,'blender-5.2.2-windows-x64')
+            self.assertEqual(module.find_blender(d,{'SPRITEMOTION_BLENDER':'custom.exe'}),'custom.exe')
+    def test_newest_runtime_beats_path_and_program_files(self):
+        with tempfile.TemporaryDirectory() as d:
+            self.make(d,'blender-5.2.2-windows-x64','blender-5.10.0-windows-x64','blender-4.2.1-windows-x64')
+            with mock.patch.object(module.shutil,'which',return_value='path-blender.exe'):
+                found=module.find_blender(d,{'ProgramFiles':d})
+            self.assertIn('blender-5.10.0',found)
+    def test_path_used_without_runtime(self):
+        with tempfile.TemporaryDirectory() as d,mock.patch.object(module.shutil,'which',return_value='path-blender.exe'):
+            self.assertEqual(module.find_blender(d,{}),'path-blender.exe')
+    def test_program_files_version_sorted_numerically(self):
+        with tempfile.TemporaryDirectory() as d,mock.patch.object(module.shutil,'which',return_value=None):
+            for v in ('Blender 4.2','Blender 10.0'):
+                exe=Path(d)/'Blender Foundation'/v/'blender.exe';exe.parent.mkdir(parents=True);exe.touch()
+            self.assertIn('Blender 10.0',module.find_blender(Path(d)/'none',{'ProgramFiles':d}))
+    def test_missing_raises(self):
+        with tempfile.TemporaryDirectory() as d,mock.patch.object(module.shutil,'which',return_value=None):
+            with self.assertRaises(RuntimeError):module.find_blender(d,{'ProgramFiles':d})
+
+class PrivateWrite(unittest.TestCase):
+    def test_content_written_and_private_mode_requested(self):
+        with tempfile.TemporaryDirectory() as d:
+            target=Path(d)/'edits.json';module.write_private(target,'{}')
+            self.assertEqual(target.read_text(),'{}')
+            if os.name!='nt':self.assertEqual(target.stat().st_mode&0o777,0o600)
+
 if __name__=='__main__':unittest.main()

tests/unit/test_jsonio.py

@@ -0,0 +1,47 @@
+"""Crash-safe JSON writes: unique temp files, and the target is never missing."""
+import os
+
+import pytest
+
+from spritemotion import jsonio
+
+
+def test_write_leaves_no_temp_files_and_round_trips(tmp_path):
+    target = tmp_path / "data.json"
+    jsonio.write_json(target, {"a": 1})
+    assert jsonio.read_json(target) == {"a": 1}
+    assert [p.name for p in tmp_path.iterdir()] == ["data.json"]
+
+
+def test_a_stale_fixed_name_tmp_file_is_not_touched(tmp_path):
+    target = tmp_path / "data.json"
+    stale = tmp_path / "data.json.tmp"
+    stale.write_text("another writer", encoding="utf-8")
+    jsonio.write_json(target, {"a": 1})
+    assert stale.read_text(encoding="utf-8") == "another writer"
+
+
+def test_target_exists_at_every_replace_with_backup(tmp_path, monkeypatch):
+    target = tmp_path / "data.json"
+    jsonio.write_json(target, {"v": 1})
+    seen = []
+    real = os.replace
+
+    def spy(src, dst):
+        seen.append(target.exists())
+        return real(src, dst)
+
+    monkeypatch.setattr(jsonio.os, "replace", spy)
+    jsonio.write_json(target, {"v": 2}, backup=True)
+    assert seen and all(seen)
+    assert jsonio.read_json(target) == {"v": 2}
+    assert jsonio.read_json(tmp_path / "data.json.bak") == {"v": 1}
+
+
+def test_a_failed_write_keeps_the_old_file_and_cleans_up(tmp_path):
+    target = tmp_path / "data.json"
+    jsonio.write_json(target, {"v": 1})
+    with pytest.raises(ValueError):
+        jsonio.write_json(target, {"v": float("nan")}, backup=True)
+    assert jsonio.read_json(target) == {"v": 1}
+    assert [p.name for p in tmp_path.iterdir()] == ["data.json"]

tests/unit/test_portability.py

@@ -0,0 +1,30 @@
+import importlib.util
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def load(rel, name):
+    spec = importlib.util.spec_from_file_location(name, ROOT / rel)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def test_load_font_falls_back_when_arial_missing(tmp_path):
+    fonts = load('games/ultima-online/region-masks/region_fonts.py', 'region_fonts')
+    font = fonts.load_font(14, paths=[tmp_path / 'missing.ttf'])
+    assert font is not None and hasattr(font, 'getmask')
+
+
+def test_pose_editor_blender_env(monkeypatch, tmp_path):
+    server = load('games/ultima-online/region-masks/pose_editor_server.py', 'pose_editor_server_t')
+    exe = tmp_path / 'blender.exe'
+    monkeypatch.setenv('SPRITEMOTION_BLENDER', str(exe))
+    assert server.find_blender() == str(exe)
+
+
+def test_uo_content_blender_env(monkeypatch, tmp_path):
+    pipeline = load('tools/uo-content/pipeline.py', 'uo_pipeline_t')
+    monkeypatch.setenv('SPRITEMOTION_BLENDER', str(tmp_path / 'b.exe'))
+    assert pipeline.blender_path() == str(tmp_path / 'b.exe')

tests/unit/test_robust_startup.py

@@ -0,0 +1,41 @@
+"""Bad annotation and status files are reported, not silently dropped or fatal."""
+import logging
+
+import pytest
+from conftest import REPO, load_module
+
+
+def test_corrupt_job_status_is_skipped_and_reported(tmp_path, monkeypatch):
+    monkeypatch.syspath_prepend(str(REPO / "tools" / "uo-content"))
+    studio = load_module(REPO / "tools" / "uo-content" / "studio.py", "uo_content_studio_test")
+    jobs = tmp_path / "jobs"
+    for name, text in (("aaaaaaaaaaaa", '{"state": "building"}'), ("bbbbbbbbbbbb", "{not json"),
+                       ("cccccccccccc", '{"nostate": 1}'), ("dddddddddddd", '{"state": "done"}')):
+        (jobs / name).mkdir(parents=True)
+        (jobs / name / "status.json").write_text(text, encoding="utf-8")
+    skipped = studio.recover_interrupted_jobs(jobs)
+    assert sorted(f.parent.name for f, _ in skipped) == ["bbbbbbbbbbbb", "cccccccccccc"]
+    assert '"failed"' in (jobs / "aaaaaaaaaaaa" / "status.json").read_text(encoding="utf-8")
+    assert (jobs / "bbbbbbbbbbbb" / "status.json").read_text(encoding="utf-8") == "{not json"
+    assert '"done"' in (jobs / "dddddddddddd" / "status.json").read_text(encoding="utf-8")
+
+
+def test_bad_annotations_log_a_warning_and_other_errors_propagate(monkeypatch, caplog):
+    pytest.importorskip("scipy")  # silhouette-fit needs it; the CI image doesn't install it
+    run = load_module(REPO / "tools" / "silhouette-fit" / "run.py", "silhouette_fit_run_test")
+
+    def bad(*args, **kwargs):
+        raise KeyError("source_fingerprint")
+
+    monkeypatch.setattr(run, "select_targets", bad)
+    with caplog.at_level(logging.WARNING, logger="silhouette-fit"):
+        targets, selection = run.annotation_targets(object(), "walk", ["head"])
+    assert targets == {} and selection["used"] == []
+    assert "source_fingerprint" in caplog.text
+
+    def broken(*args, **kwargs):
+        raise RuntimeError("programming error")
+
+    monkeypatch.setattr(run, "select_targets", broken)
+    with pytest.raises(RuntimeError):
+        run.annotation_targets(object(), "walk", ["head"])