mirror of
https://github.com/thunderheronight/splat-workshop.git
synced 2026-10-17 13:02:07 +02:00
0.2.1: automated tests and GitHub Actions; a cancelled job only ends with its last process
57 tests without GPU, recordings or Frame: fake pipeline, fake splat-transform and trash, and synthetic rooms with a known error for putting rooms together. Found on the way: after cancelling, the job counted as finished when systemd-inhibit ended, while splat.sh was still putting the previous scene back – pid_alive now looks at the whole process group. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
thunderheronightandClaude Opus 5.5 committed
1 parent
49124eacc8
commit
a8a6aa5188
15 files changed
+909
-5
No files matched your search
@@ -0,0 +1,14 @@
|
||||
name: Tests
|
||||
|
||||
on:
|
||||
push:
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
tests:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- uses: astral-sh/setup-uv@v6
|
||||
- name: Run the tests
|
||||
run: uv run --no-project --python 3.12 --with pytest --with numpy --with scipy --with pillow pytest -v
|
||||
@@ -3,3 +3,4 @@ __pycache__/
|
||||
.venv/
|
||||
.splat-workshop/
|
||||
*.part
|
||||
.pytest_cache/
|
||||
@@ -2,6 +2,16 @@
|
||||
|
||||
All notable changes to Splat Workshop. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), versions follow [Semantic Versioning](https://semver.org/).
|
||||
|
||||
## [0.2.1] – 2026-10-14
|
||||
|
||||
### Added
|
||||
|
||||
- **Automated tests** (57) that run on every push via GitHub Actions – no GPU, recordings or Frame needed: the server's protection against other websites and malformed requests, scenes, groups, conversions with cancel/start over/resume, tours, the four corrections on synthetic rooms with a known error, borders through doors, and the floor plan's orientation. See *Development* in the README.
|
||||
|
||||
### Fixed
|
||||
|
||||
- After cancelling, a conversion counted as finished as soon as the sleep inhibitor ended, while `splat.sh` was still putting the previous scene back – now the whole job has to end.
|
||||
|
||||
## [0.2.0] – 2026-10-14
|
||||
|
||||
The big one: **put the rooms of your home together into one walkable tour.**
|
||||
@@ -44,5 +54,6 @@ First release.
|
||||
- **Floor plans** (experimental) in two styles with a measuring tool.
|
||||
- Configurable video player; English and German interface.
|
||||
|
||||
[0.2.1]: https://github.com/thunderheronight/splat-workshop/compare/v0.2.0...v0.2.1
|
||||
[0.2.0]: https://github.com/thunderheronight/splat-workshop/compare/v0.1.0...v0.2.0
|
||||
[0.1.0]: https://github.com/thunderheronight/splat-workshop/releases/tag/v0.1.0
|
||||
+1
-1
@@ -2,7 +2,7 @@
|
||||
|
||||
**Eine grafische Oberfläche für die Rekonstruktions-Pipeline von Arcturus Vision: Aufnahme der Arcturus Vision Camera an der Steam Frame auswählen, prüfen, umwandeln lassen und das Ergebnis direkt in die Gaussian-XR-Galerie auf der Brille bringen. Mit Weboberfläche, Tray-Symbol und Aufnahmeprüfung.**
|
||||
|
||||
  [](https://github.com/thunderheronight/splat-workshop/releases)
|
||||
  [](https://github.com/thunderheronight/splat-workshop/releases) [](https://github.com/thunderheronight/splat-workshop/actions/workflows/tests.yml)
|
||||
|
||||
> **Kein Root, kein Jailbreak.** Auf der Frame nutzt die Splat-Werkstatt nur den offiziellen *Entwicklermodus* aus den Steam-Einstellungen und SSH als normaler Benutzer – auf der Brille wird nichts installiert, ihr System bleibt unangetastet. Auch am PC läuft sie als dein Benutzer (nur die Voraussetzungen wie ffmpeg kommen aus der Paketverwaltung).
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
**A graphical interface for the Arcturus Vision reconstruction pipeline: pick a recording from the Arcturus Vision Camera on your Steam Frame, check it, run the conversion and get the result straight into the Gaussian XR gallery on the headset. With a web interface, a tray icon and a recording check.**
|
||||
|
||||
  [](https://github.com/thunderheronight/splat-workshop/releases)
|
||||
  [](https://github.com/thunderheronight/splat-workshop/releases) [](https://github.com/thunderheronight/splat-workshop/actions/workflows/tests.yml)
|
||||
|
||||
> **No root, no jailbreak.** On the Frame, Splat Workshop only uses the official *Developer Mode* from the Steam settings and SSH as the normal user – nothing is installed on the headset and its system stays untouched. On the PC it runs as your user, too (only the prerequisites like ffmpeg come from your package manager).
|
||||
|
||||
@@ -165,6 +165,16 @@ tours: [paths] runs/<group>/ (scene.sog = Frame version, full/ = ful
|
||||
internal state: next to runs/ in .splat-workshop/ (logs, thumbnails, check results)
|
||||
```
|
||||
|
||||
## Development
|
||||
|
||||
Automated tests run on every push ([GitHub Actions](https://github.com/thunderheronight/splat-workshop/actions/workflows/tests.yml)). Run them yourself with:
|
||||
|
||||
```bash
|
||||
uv run --no-project --with pytest --with numpy --with scipy --with pillow pytest
|
||||
```
|
||||
|
||||
They need no GPU, no recordings and no Steam Frame: the Arcturus pipeline, Spirula, splat-transform and the desktop trash are replaced by small fakes, and putting rooms together is tested on synthetic rooms with a known error (`tests/synth.py`). Covered: who may talk to the server (other websites, LAN mode, malformed requests, paths), renaming, clean-up, deleting, cancelled runs, groups, conversions with cancel/start over/resume (the real `splat.sh`), tours (the real `combine.sh`), the four corrections, borders through doors, and that the floor plan is not mirrored. Not covered: the real pipeline, Spirula and the Frame – those are tested by hand.
|
||||
|
||||
## Uninstall
|
||||
|
||||
`./install.sh --uninstall` removes the menu entry, icon and autostart. Your config and scenes stay.
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
[pytest]
|
||||
testpaths = tests
|
||||
@@ -29,7 +29,7 @@ from pathlib import Path
|
||||
from . import check, config
|
||||
from .i18n import T, detect
|
||||
|
||||
VERSION = "0.2.0"
|
||||
VERSION = "0.2.1"
|
||||
CFG = config.load()
|
||||
ROOT = config.ROOT
|
||||
WEB = ROOT / "web"
|
||||
@@ -134,10 +134,22 @@ def save_state(st: dict) -> None:
|
||||
|
||||
|
||||
def pid_alive(pid: int) -> bool:
|
||||
"""Is the job started as `pid` still running? The job is its whole process group: the first process
|
||||
(systemd-inhibit, if the PC is kept awake) can end before splat.sh has finished cleaning up."""
|
||||
if proc is not None and proc.pid == pid:
|
||||
proc.poll() # reap it, so a finished first process does not count
|
||||
try:
|
||||
cmd = Path(f"/proc/{pid}/cmdline").read_bytes().decode(errors="replace")
|
||||
return "splat.sh" in cmd or "combine.sh" in cmd
|
||||
if cmd:
|
||||
return "splat.sh" in cmd or "combine.sh" in cmd
|
||||
except OSError:
|
||||
pass
|
||||
if not pid:
|
||||
return False
|
||||
try:
|
||||
os.killpg(pid, 0) # the first process is gone, are others of the job left?
|
||||
return True
|
||||
except (ProcessLookupError, PermissionError):
|
||||
return False
|
||||
|
||||
|
||||
@@ -503,7 +515,7 @@ def job_status(tail=True) -> dict | None:
|
||||
return None
|
||||
out = Path(st["out"])
|
||||
text = out.read_text(errors="replace") if out.exists() else ""
|
||||
running = (proc is not None and proc.poll() is None) or (proc is None and pid_alive(st["pid"]))
|
||||
running = pid_alive(st["pid"])
|
||||
parsed = parse_output(text)
|
||||
res = {**st, **parsed, "running": running}
|
||||
if tail:
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
"""Test setup: a temporary home for Splat Workshop with fake tools, and the server on a free port.
|
||||
|
||||
The real Arcturus pipeline, Spirula, splat-transform and the desktop trash are replaced by small fakes
|
||||
(written below), so the tests run anywhere – no GPU, no recordings, no Steam Frame. The server reads
|
||||
its configuration when it is imported, so everything is set up before that, once per test session.
|
||||
"""
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path[:0] = [str(ROOT), str(Path(__file__).parent)]
|
||||
|
||||
BASE = Path(tempfile.mkdtemp(prefix="splat-workshop-tests-"))
|
||||
RUNS, REC, STATE, BIN, TRASH, PIPE = (BASE / n for n in ("runs", "recordings", "state", "bin", "trash", "pipeline"))
|
||||
for d in (RUNS, REC, STATE, BIN, TRASH, PIPE):
|
||||
d.mkdir()
|
||||
|
||||
|
||||
def script(path: Path, text: str):
|
||||
path.write_text(text)
|
||||
path.chmod(0o755)
|
||||
|
||||
|
||||
# Fake pipeline: behaves like av1_to_3dgs.py as far as Splat Workshop can see (progress lines, result files).
|
||||
# FAKE_PIPELINE=ok (default) | fail | slow (waits for 60 s, to be cancelled)
|
||||
script(PIPE / "av1_to_3dgs.py", r'''
|
||||
import base64, json, os, sys, time
|
||||
from pathlib import Path
|
||||
out = Path(sys.argv[2])
|
||||
mode = os.environ.get("FAKE_PIPELINE", "ok")
|
||||
(out / "dataset").mkdir(parents=True, exist_ok=True)
|
||||
print("Extracting 20 images from 2 video streams", flush=True)
|
||||
(out / "dataset" / "spirula-dataset.json").write_text(json.dumps({"status": "complete"}))
|
||||
print("Running native Spirula SfM", flush=True)
|
||||
print("Aligned 20 cameras using orientation anchor cam0/0000.jpg: RMS 0.0110 m, median 0.0090 m.", flush=True)
|
||||
if mode == "fail":
|
||||
sys.exit("fake pipeline: failed on purpose")
|
||||
for step in (1000, 2000, 3000):
|
||||
print(f"step {step:6d}/3000 ({step // 30:3d}%) splats 1000 [elapsed 00:00:0{step // 1000}.000 | ETA 00:00:01.000] rgb_loss=0.1 ssim=0.8 psnr=28.5", flush=True)
|
||||
if mode == "slow":
|
||||
time.sleep(20)
|
||||
(out / "spirula").mkdir(exist_ok=True)
|
||||
(out / "spirula" / "scene.ply").write_bytes(b"ply\n")
|
||||
print("spirula: 1,000 Gaussians in " + str(out / "spirula" / "scene.ply"), flush=True)
|
||||
(out / "scene.sog").write_bytes(b"fake sog")
|
||||
(out / "scene.xrSceneTransform.json").write_text(json.dumps({"transform": [[1, 0, 0, 0], [0, 1, 0, 0], [0, 0, 1, 0], [0, 0, 0, 1]]}))
|
||||
png = base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==")
|
||||
(out / (out.name.lower() + ".gallery.png")).write_bytes(png)
|
||||
print("Gaussian XR scene transform: " + str(out / "scene.xrSceneTransform.json"), flush=True)
|
||||
''')
|
||||
script(BIN / "spirula", "#!/bin/sh\nexit 0\n")
|
||||
# splat-transform: copies the input (a .ply stays a .ply), writes a small file for .sog
|
||||
script(BIN / "splat-transform", f'''#!{sys.executable}
|
||||
import shutil, sys
|
||||
args = [a for a in sys.argv[1:] if a not in ("-w", "--overwrite")]
|
||||
if "--decimate" in args:
|
||||
i = args.index("--decimate"); del args[i:i + 2]
|
||||
src, dst = args[0], args[-1]
|
||||
if dst.endswith(".ply"):
|
||||
shutil.copy(src, dst)
|
||||
else:
|
||||
open(dst, "wb").write(b"fake sog")
|
||||
print("done")
|
||||
''')
|
||||
# gio trash: moves into a trash folder of the test (numbered if a name repeats)
|
||||
script(BIN / "gio", f'''#!/bin/sh
|
||||
[ "$1" = trash ] || exit 1
|
||||
shift; [ "$1" = -- ] && shift
|
||||
exec mv --backup=numbered -t "{TRASH}" -- "$@"
|
||||
''')
|
||||
|
||||
(BASE / "config.toml").write_text(f'''
|
||||
[paths]
|
||||
pipeline = "{PIPE}"
|
||||
spirula = "{BIN / 'spirula'}"
|
||||
splat_transform = "{BIN / 'splat-transform'}"
|
||||
recordings = "{REC}"
|
||||
runs = "{RUNS}"
|
||||
extra_recording_dirs = []
|
||||
[server]
|
||||
language = "en"
|
||||
[frame]
|
||||
host = ""
|
||||
''')
|
||||
os.environ.update(SPLAT_WORKSHOP_CONFIG=str(BASE / "config.toml"), SPLAT_WORKSHOP_STATE=str(STATE), FRAME_HOST="",
|
||||
PATH=f"{BIN}:{os.environ['PATH']}")
|
||||
os.environ.pop("SPLAT_LAN", None)
|
||||
|
||||
from splat_workshop import server # noqa: E402 (after the environment is set up)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def clean():
|
||||
"""Every test starts with empty runs, recordings, state and trash."""
|
||||
yield
|
||||
st = server.load_state().get("job") or {}
|
||||
if server.pid_alive(st.get("pid", 0)):
|
||||
server.cancel_job()
|
||||
wait_job(timeout=20)
|
||||
for d in (RUNS, REC, STATE, TRASH):
|
||||
shutil.rmtree(d, ignore_errors=True)
|
||||
d.mkdir()
|
||||
for cache in (server.usage_cache, server.probe_cache, server.uploads, server.checks):
|
||||
cache.clear()
|
||||
os.environ.pop("FAKE_PIPELINE", None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ok_probe(monkeypatch):
|
||||
"""Recordings are fake files: let the recording check (ffprobe) accept them."""
|
||||
monkeypatch.setattr(server, "probe", lambda path: {"ok": True, "errors": [], "warnings": [], "duration": 60})
|
||||
|
||||
|
||||
def recording(name="2026-10-12_16-40-00_stereo.mp4"):
|
||||
p = REC / name
|
||||
p.write_bytes(b"not really a video")
|
||||
return p
|
||||
|
||||
|
||||
def wait_job(timeout=120):
|
||||
"""Wait until the current job has ended; returns its status."""
|
||||
end = time.time() + timeout
|
||||
while time.time() < end:
|
||||
js = server.job_status(tail=True)
|
||||
if js and not js["running"]:
|
||||
return js
|
||||
time.sleep(0.3)
|
||||
raise TimeoutError("job did not finish")
|
||||
|
||||
|
||||
class Client:
|
||||
def __init__(self, port):
|
||||
self.port = port
|
||||
|
||||
def request(self, method, path, body=None, headers=None):
|
||||
c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=30)
|
||||
data = body if isinstance(body, (bytes, type(None))) else json.dumps(body).encode()
|
||||
c.request(method, path, body=data, headers={"Content-Type": "application/json", **(headers or {})})
|
||||
r = c.getresponse()
|
||||
raw = r.read()
|
||||
try:
|
||||
return r.status, json.loads(raw)
|
||||
except ValueError:
|
||||
return r.status, raw
|
||||
|
||||
def get(self, path, headers=None):
|
||||
return self.request("GET", path, headers=headers)
|
||||
|
||||
def post(self, path, body=None, headers=None):
|
||||
return self.request("POST", path, body if body is not None else {}, headers)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def api():
|
||||
srv = server.http.server.ThreadingHTTPServer(("127.0.0.1", 0), server.Handler)
|
||||
srv.daemon_threads = True
|
||||
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
||||
yield Client(srv.server_address[1])
|
||||
srv.shutdown()
|
||||
|
||||
|
||||
def pytest_sessionfinish(session, exitstatus):
|
||||
shutil.rmtree(BASE, ignore_errors=True)
|
||||
+163
@@ -0,0 +1,163 @@
|
||||
"""Synthetic rooms in the shape the Arcturus pipeline leaves them, for tests without real recordings.
|
||||
|
||||
Two rooms side by side with a door between them (Y points down like in the pipeline):
|
||||
|
||||
z=3 +----------------+----------------+
|
||||
| | |
|
||||
| room A door room B |
|
||||
| (x 0..4) | (x 4..8) |
|
||||
z=0 +----------------+----------------+
|
||||
x=0 x=4 x=8
|
||||
|
||||
Every recording walks around its room; room B starts in room A next to the door and walks in
|
||||
(like a real recording that starts in the hallway). Each room gets what the pipeline writes:
|
||||
dataset/main.aiscene camera poses as tracked by the Frame (the truth, plus `drift`)
|
||||
dataset/sparse/0/images.bin camera poses of the reconstruction
|
||||
dataset/sparse/0/points3D.bin sparse points
|
||||
spirula/scene.ply splats (one per point)
|
||||
scene.sog, scene.xrSceneTransform.json, <slug>.ui.json
|
||||
The reconstruction is off by a known small rotation and shift (`error`), as a real one is before
|
||||
it is fitted to the Frame's tracking.
|
||||
"""
|
||||
import json
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
from scipy.spatial.transform import Rotation
|
||||
|
||||
FLOOR, CEILING = 1.6, -0.9 # y (down): floor below the cameras, ceiling above
|
||||
DOOR = (1.2, 2.0) # door between the rooms, z range
|
||||
PROPS = (["x", "y", "z", "nx", "ny", "nz", "f_dc_0", "f_dc_1", "f_dc_2"] + [f"f_rest_{i}" for i in range(45)]
|
||||
+ ["opacity", "scale_0", "scale_1", "scale_2", "rot_0", "rot_1", "rot_2", "rot_3"])
|
||||
|
||||
|
||||
def walls(x0, x1, step=0.04):
|
||||
"""Points on the walls, floor and ceiling of the room x0..x1 (z 0..3), with a door in the wall at x=4."""
|
||||
pts = []
|
||||
ys = np.arange(CEILING, FLOOR + 1e-9, step)
|
||||
for x in (x0, x1):
|
||||
for z in np.arange(0, 3 + 1e-9, step):
|
||||
if x == 4 and DOOR[0] < z < DOOR[1]:
|
||||
continue
|
||||
pts += [(x, y, z) for y in ys]
|
||||
for z in (0, 3):
|
||||
for x in np.arange(x0, x1 + 1e-9, step):
|
||||
pts += [(x, y, z) for y in ys]
|
||||
for x in np.arange(x0, x1 + 1e-9, step * 2):
|
||||
for z in np.arange(0, 3 + 1e-9, step * 2):
|
||||
pts += [(x, FLOOR, z), (x, CEILING, z)]
|
||||
return np.array(pts)
|
||||
|
||||
|
||||
def path(room):
|
||||
"""Camera positions over time (y = 0, head height) and the times in seconds."""
|
||||
t = np.linspace(0, 2 * np.pi, 140)
|
||||
if room == "A":
|
||||
# starts at the far wall, passes the door in the middle of the round
|
||||
loop = np.stack([2 + 1.3 * np.cos(t + np.pi), np.zeros_like(t), 1.5 + 0.9 * np.sin(t + np.pi)], 1)
|
||||
return loop, np.linspace(0, 40, len(loop))
|
||||
# starts on room A's round (like a recording that starts in the hallway), then through the door
|
||||
tt = np.linspace(-1.0, 0, 18)
|
||||
# 30 cm closer to room A's wall than room A's own recording went
|
||||
along_a = np.stack([2 + 1.3 * np.cos(tt), np.zeros_like(tt), 1.5 + 0.9 * np.sin(tt) - 0.3], 1)
|
||||
walk_in = np.vstack([along_a, np.stack([np.linspace(3.35, 4.6, 12), np.zeros(12), np.full(12, 1.6)], 1)])
|
||||
loop = np.stack([6 + 1.3 * np.cos(t + np.pi), np.zeros_like(t), 1.5 + 0.9 * np.sin(t + np.pi)], 1)
|
||||
pos = np.vstack([walk_in, loop])
|
||||
return pos, np.linspace(0, 45, len(pos))
|
||||
|
||||
|
||||
def look_at(c, target):
|
||||
"""cam_from_world rotation for a camera at c looking at target (x right, y down, z forward)."""
|
||||
z = target - c
|
||||
z[1] = 0
|
||||
z /= np.linalg.norm(z)
|
||||
y = np.array([0.0, 1.0, 0.0])
|
||||
x = np.cross(y, z)
|
||||
return np.stack([x, y, z])
|
||||
|
||||
|
||||
def write_images(path_, names, R_list, C_list):
|
||||
with open(path_, "wb") as f:
|
||||
f.write(struct.pack("<Q", len(names)))
|
||||
for i, (name, R, C) in enumerate(zip(names, R_list, C_list)):
|
||||
qx, qy, qz, qw = Rotation.from_matrix(R).as_quat()
|
||||
t = -R @ C
|
||||
f.write(struct.pack("<i7di", i + 1, qw, qx, qy, qz, *t, 1))
|
||||
f.write(name.encode() + b"\0")
|
||||
f.write(struct.pack("<Q", 0))
|
||||
|
||||
|
||||
def write_points(path_, pts):
|
||||
with open(path_, "wb") as f:
|
||||
f.write(struct.pack("<Q", len(pts)))
|
||||
for i, p in enumerate(pts):
|
||||
f.write(struct.pack("<Q3d3BdQ", i + 1, *p, 128, 128, 128, 0.5, 0))
|
||||
|
||||
|
||||
def write_ply(path_, pts, colours):
|
||||
data = np.zeros(len(pts), dtype=np.dtype([(p, "<f4") for p in PROPS]))
|
||||
data["x"], data["y"], data["z"] = pts.T
|
||||
for k in range(3):
|
||||
data[f"f_dc_{k}"] = (colours[:, k] - 0.5) / 0.28209479177387814
|
||||
data["opacity"] = 3.0
|
||||
for k in range(3):
|
||||
data[f"scale_{k}"] = np.log(0.02)
|
||||
data["rot_0"] = 1.0
|
||||
header = "ply\nformat binary_little_endian 1.0\nelement vertex %d\n%send_header\n" % (
|
||||
len(pts), "".join(f"property float {p}\n" for p in PROPS))
|
||||
with open(path_, "wb") as f:
|
||||
f.write(header.encode())
|
||||
data.tofile(f)
|
||||
|
||||
|
||||
def make_room(runs: Path, slug: str, room: str, error=(1.0, (0.03, 0.0, -0.02)), drift=(0.0, 0.0, 0.0),
|
||||
name=None, recording=None, markers=False, colour=None):
|
||||
"""Write one synthetic room into runs/<slug>. error = (yaw in degrees, shift in m) of the reconstruction;
|
||||
drift = constant offset of the Frame's tracking for this recording. Returns the run folder."""
|
||||
run = runs / slug
|
||||
sparse = run / "dataset" / "sparse" / "0"
|
||||
sparse.mkdir(parents=True, exist_ok=True)
|
||||
(run / "spirula").mkdir(exist_ok=True)
|
||||
x0, x1 = (0.0, 4.0) if room == "A" else (4.0, 8.0)
|
||||
pts = walls(x0, x1)
|
||||
# seen through the door: a little of the neighbouring room close to the door
|
||||
other = walls(4.0, 8.0) if room == "A" else walls(0.0, 4.0)
|
||||
near_door = other[(np.abs(other[:, 0] - 4) < 1.0) & (other[:, 2] > 0.6) & (other[:, 2] < 2.6)]
|
||||
pts = np.vstack([pts, near_door])
|
||||
# every recording its own colour, so the origin of a splat can be seen after putting rooms together
|
||||
colours = np.tile(colour or ((0.8, 0.3, 0.3) if room == "A" else (0.3, 0.3, 0.8)), (len(pts), 1))
|
||||
if markers: # red at the origin corner, green along +x, blue along +z
|
||||
for c, at in (((1, 0, 0), (0.0, 0.5, 0.0)), ((0, 1, 0), (2.0, 0.5, 0.0)), ((0, 0, 1), (0.0, 0.5, 2.0))):
|
||||
blob = np.array(at) + np.random.default_rng(1).normal(0, 0.05, (400, 3))
|
||||
pts = np.vstack([pts, blob])
|
||||
colours = np.vstack([colours, np.tile(c, (400, 1))])
|
||||
|
||||
pos, times = path(room)
|
||||
centre = np.array([(x0 + x1) / 2, 0.0, 1.5])
|
||||
yaw, shift = error
|
||||
E = Rotation.from_euler("y", yaw, degrees=True).as_matrix() # world -> reconstruction
|
||||
to_sfm = lambda p: (E @ p.T).T + np.array(shift)
|
||||
names, Rs, Cs, frames = [], [], [], []
|
||||
for k, (c, t) in enumerate(zip(pos, times)):
|
||||
R_world = look_at(c.copy(), centre)
|
||||
image = f"cam0/{k:04d}-{t:.6f}.jpg"
|
||||
names.append(image)
|
||||
Rs.append(R_world @ E.T)
|
||||
Cs.append(to_sfm(c))
|
||||
world_from_cam = np.eye(4)
|
||||
world_from_cam[:3, :3] = R_world.T
|
||||
world_from_cam[:3, 3] = c + np.array(drift)
|
||||
frames.append({"cam": 0, "path": "images/" + image, "worldFromCam": world_from_cam.tolist(),
|
||||
"playbackTimestampSec": float(t)})
|
||||
write_images(sparse / "images.bin", names, Rs, Cs)
|
||||
write_points(sparse / "points3D.bin", to_sfm(pts))
|
||||
write_ply(run / "spirula" / "scene.ply", to_sfm(pts), colours)
|
||||
(run / "dataset" / "main.aiscene").write_text(json.dumps({"format": "synthetic", "cameras": [], "frames": frames}))
|
||||
(run / "scene.sog").write_bytes(b"synthetic")
|
||||
(run / "scene.xrSceneTransform.json").write_text(json.dumps({"transform": np.eye(4).tolist(), "worldScale": 1.0}))
|
||||
(runs / f"{slug}.log").write_text(f"spirula: {len(pts):,} Gaussians in {run}/spirula/scene.ply\n")
|
||||
meta = {"name": name or slug, "recording": recording or f"2026-10-12_16-{40 + (room == 'B') * 3:02d}-00_stereo.mp4",
|
||||
"quality": "high"}
|
||||
(runs / f"{slug}.ui.json").write_text(json.dumps(meta))
|
||||
return run
|
||||
@@ -0,0 +1,101 @@
|
||||
"""Putting rooms together (scripts/combine.py) on synthetic rooms with a known error."""
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
from scipy.spatial.transform import Rotation
|
||||
|
||||
from conftest import ROOT, RUNS, BASE
|
||||
import synth
|
||||
|
||||
COMBINE = ROOT / "scripts" / "combine.py"
|
||||
YAW_A, YAW_B = 1.0, -0.8 # degrees: how far each reconstruction is turned
|
||||
DRIFT_B = 0.03 # metres: the Frame's tracking of room B is off by this much
|
||||
|
||||
|
||||
def two_rooms(drift_b=DRIFT_B, b_far=False):
|
||||
a = synth.make_room(RUNS, "A", "A", error=(YAW_A, (0.03, 0.0, -0.02)))
|
||||
b = synth.make_room(RUNS, "B", "B", error=(YAW_B, (-0.02, 0.01, 0.03)),
|
||||
drift=(20.0 if b_far else drift_b, 0.0, 0.0))
|
||||
return a, b
|
||||
|
||||
|
||||
def combine(*args, align="images", partition="view"):
|
||||
out = BASE / f"out-{align}-{partition}"
|
||||
r = subprocess.run([sys.executable, str(COMBINE), str(out), *[f"{p}={p.name}" for p in args],
|
||||
"--align", align, "--partition", partition], capture_output=True, text=True)
|
||||
return out, r
|
||||
|
||||
|
||||
def result(out):
|
||||
return json.loads((out / "combine.json").read_text())
|
||||
|
||||
|
||||
def read_ply(path):
|
||||
with open(path, "rb") as f:
|
||||
header = b""
|
||||
while not header.endswith(b"end_header\n"):
|
||||
header += f.readline()
|
||||
props = [l.split()[-1].decode() for l in header.splitlines() if l.startswith(b"property")]
|
||||
n = int(next(l for l in header.splitlines() if l.startswith(b"element vertex")).split()[-1])
|
||||
return np.fromfile(f, dtype=np.dtype([(p, "<f4") for p in props]), count=n)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("align,lo,hi", [
|
||||
("frame", 5.0, 12.0), # as delivered: both reconstructions' errors add up
|
||||
("images", 2.0, 4.0), # rooms fitted, but room B's tracking drift (3 cm) is left
|
||||
("shift", 0.0, 0.5), # shifting the rooms over the door removes the drift
|
||||
("rigid", 0.0, 0.5),
|
||||
])
|
||||
def test_corrections_take_out_the_known_error(align, lo, hi):
|
||||
out, r = combine(*two_rooms(), align=align)
|
||||
assert r.returncode == 0, r.stderr
|
||||
seam = result(out)["seams"]["A|B"]["cm"]
|
||||
assert lo <= seam <= hi, f"{align}: overlap off by {seam} cm"
|
||||
|
||||
|
||||
def test_every_room_is_fitted_to_the_frames_tracking():
|
||||
out, r = combine(*two_rooms(drift_b=0.0))
|
||||
rooms = result(out)["rooms"]
|
||||
for room in rooms.values():
|
||||
assert room["tracking_cm"] > 3 # off before: turned and shifted
|
||||
assert room["fitted_cm"] < 0.1 # fitted over all images: exact
|
||||
# the splats are turned back as well (they start without rotation in the synthetic rooms)
|
||||
v = read_ply(out / "spirula" / "scene.ply")
|
||||
q = np.stack([v["rot_1"], v["rot_2"], v["rot_3"], v["rot_0"]], 1)
|
||||
yaws = np.abs(Rotation.from_quat(q).as_euler("yxz", degrees=True)[:, 0])
|
||||
assert np.isclose(sorted(set(np.round(yaws, 1))), [abs(YAW_B), YAW_A], atol=0.05).all()
|
||||
|
||||
|
||||
def test_all_splats_are_written_once():
|
||||
out, r = combine(*two_rooms())
|
||||
info = result(out)
|
||||
v = read_ply(out / "spirula" / "scene.ply")
|
||||
assert len(v) == info["splats"] == sum(room["kept"] for room in info["rooms"].values())
|
||||
assert (out / "scene.xrSceneTransform.json").exists()
|
||||
|
||||
|
||||
def test_borders_run_through_the_door():
|
||||
"""Room B starts its recording in room A (walking in). With borders 'through doors' that stretch
|
||||
claims nothing, so room A comes completely from room A's own recording."""
|
||||
def colours_in_room_a(partition):
|
||||
out, r = combine(*two_rooms(drift_b=0.0), partition=partition)
|
||||
assert r.returncode == 0, r.stderr
|
||||
v = read_ply(out / "spirula" / "scene.ply")
|
||||
inside_a = (v["x"] > 0.2) & (v["x"] < 3.6)
|
||||
blue = v["f_dc_2"] > v["f_dc_0"] # room B's splats are blue, room A's red
|
||||
return int((inside_a & blue).sum()), result(out)["rooms"]["B"]["walk_in_out_cameras"]
|
||||
from_b_view, walk_in = colours_in_room_a("view")
|
||||
from_b_nearest, _ = colours_in_room_a("nearest")
|
||||
assert walk_in > 10 # recognised as walking in
|
||||
assert from_b_view < 50 # practically nothing of room B inside room A
|
||||
# nearest camera: room B's walk-in cuts into room A's wall
|
||||
assert from_b_nearest > 50 and from_b_nearest > 10 * max(from_b_view, 1)
|
||||
|
||||
|
||||
def test_rooms_from_different_sessions_are_refused():
|
||||
out, r = combine(*two_rooms(b_far=True))
|
||||
assert r.returncode != 0
|
||||
assert "@@ error not_connected" in r.stdout
|
||||
@@ -0,0 +1,46 @@
|
||||
"""The floor plan is a top view, not its mirror image (scripts/floorplan.py).
|
||||
|
||||
A plan may be turned (so the walls are straight), but never mirrored. Three coloured markers tell
|
||||
the two apart: red at the corner, green along +x, blue along +z. Turning keeps the sense of
|
||||
red -> green -> blue, mirroring reverses it. Seen from above with the y axis pointing down
|
||||
(like the pipeline's frame), +x to +z turns counter-clockwise on screen.
|
||||
"""
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
from conftest import BASE, ROOT
|
||||
import synth
|
||||
|
||||
|
||||
def draw(style):
|
||||
runs = BASE / f"plan-{style}"
|
||||
synth.make_room(runs, "Room", "A", error=(0.0, (0, 0, 0)), markers=True)
|
||||
out = runs / "plan.png"
|
||||
r = subprocess.run([sys.executable, str(ROOT / "scripts" / "floorplan.py"), str(out), f"{runs / 'Room'}=Room",
|
||||
"--style", style], capture_output=True, text=True)
|
||||
assert r.returncode == 0, r.stderr
|
||||
return np.asarray(Image.open(out).convert("RGB")).astype(int)
|
||||
|
||||
|
||||
def test_floor_plan_is_not_mirrored():
|
||||
img = draw("pretty") # the pretty style shows the splats' own colours
|
||||
rr, gg, bb = img[..., 0], img[..., 1], img[..., 2]
|
||||
|
||||
def centre(mask):
|
||||
ys, xs = np.nonzero(mask)
|
||||
assert len(xs) > 50, "marker not found"
|
||||
return np.array([xs.mean(), ys.mean()])
|
||||
red = centre((rr > 150) & (gg < 90) & (bb < 90))
|
||||
green = centre((gg > 150) & (rr < 90) & (bb < 90))
|
||||
blue = centre((bb > 150) & (rr < 90) & (gg < 90))
|
||||
gx, gz = green - red, blue - red # screen directions of +x and +z (rows count downwards)
|
||||
cross = gx[0] * gz[1] - gx[1] * gz[0]
|
||||
assert cross < 0, "the floor plan is mirrored"
|
||||
|
||||
|
||||
def test_classic_style_draws():
|
||||
img = draw("classic")
|
||||
assert img.shape[0] > 500 and img.std() > 5 # not an empty picture
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Groups of scenes: one group per scene, suggestions, renaming, cleaning up together."""
|
||||
import json
|
||||
|
||||
from conftest import RUNS, server
|
||||
import synth
|
||||
|
||||
|
||||
def make_flat():
|
||||
synth.make_room(RUNS, "Kitchen", "A", name="Kitchen", recording="2026-10-12_16-40-00_stereo.mp4")
|
||||
synth.make_room(RUNS, "Hallway", "B", name="Hallway", recording="2026-10-12_16-43-00_stereo.mp4")
|
||||
synth.make_room(RUNS, "Statue", "A", name="Statue", recording="2026-10-03_19-00-00_stereo.mp4")
|
||||
|
||||
|
||||
def test_rooms_recorded_one_after_another_are_suggested(api):
|
||||
make_flat()
|
||||
status, r = api.get("/api/groups")
|
||||
assert status == 200 and r["groups"] == []
|
||||
assert sorted(r["suggestion"]["members"]) == ["Hallway", "Kitchen"] # not the statue from another day
|
||||
assert r["suggestion"]["date"] == "2026-10-12"
|
||||
assert api.post("/api/group/dismiss", {"key": r["suggestion"]["key"]})[0] == 200
|
||||
assert api.get("/api/groups")[1]["suggestion"] is None
|
||||
|
||||
|
||||
def test_a_group_with_its_scenes(api):
|
||||
make_flat()
|
||||
status, r = api.post("/api/group", {"name": "Example home", "members": "Kitchen,Hallway"})
|
||||
assert status == 200 and r["slug"] == "Example-home"
|
||||
g = api.get("/api/groups")[1]["groups"][0]
|
||||
assert (g["name"], g["members"], g["result"]) == ("Example home", ["Kitchen", "Hallway"], None)
|
||||
assert "Example-home" not in {s["slug"] for s in api.get("/api/scenes")[1]} # a group is not a scene
|
||||
assert "Example-home" not in {u["slug"] for u in api.get("/api/unfinished")[1]}
|
||||
|
||||
|
||||
def test_a_scene_can_only_be_in_one_group(api):
|
||||
make_flat()
|
||||
api.post("/api/group", {"name": "Home", "members": "Kitchen,Hallway"})
|
||||
status, r = api.post("/api/group", {"name": "Other", "members": "Kitchen,Statue"})
|
||||
assert status == 400 and r["error"]["code"] == "err.in_other_group"
|
||||
assert api.post("/api/group", {"name": "Empty", "members": ""})[0] == 400
|
||||
assert api.post("/api/group", {"name": "Ghost", "members": "Nope"})[0] == 400
|
||||
|
||||
|
||||
def test_renaming_and_changing_a_group(api):
|
||||
make_flat()
|
||||
api.post("/api/group", {"name": "Home", "members": "Kitchen,Hallway"})
|
||||
status, r = api.post("/api/group", {"slug": "Home", "name": "Old flat", "members": "Kitchen"})
|
||||
assert status == 200 and r["slug"] == "Old-flat"
|
||||
assert not (RUNS / "Home.ui.json").exists()
|
||||
g = api.get("/api/groups")[1]["groups"][0]
|
||||
assert (g["slug"], g["name"], g["members"]) == ("Old-flat", "Old flat", ["Kitchen"])
|
||||
|
||||
|
||||
def test_renamed_and_deleted_scenes_follow_their_group(api):
|
||||
make_flat()
|
||||
api.post("/api/group", {"name": "Home", "members": "Kitchen,Hallway"})
|
||||
api.post("/api/rename", {"slug": "Kitchen", "new": "Küche"})
|
||||
assert json.loads((RUNS / "Home.ui.json").read_text())["members"] == ["Kueche", "Hallway"]
|
||||
api.post("/api/delete", {"slug": "Hallway"})
|
||||
assert json.loads((RUNS / "Home.ui.json").read_text())["members"] == ["Kueche"]
|
||||
|
||||
|
||||
def test_cleaning_up_a_group_cleans_only_its_scenes(api):
|
||||
make_flat()
|
||||
for slug in ("Kitchen", "Hallway", "Statue"):
|
||||
(RUNS / slug / "spirula" / "training").mkdir(parents=True)
|
||||
(RUNS / slug / "spirula" / "training" / "x").write_bytes(b"x" * 1000)
|
||||
api.post("/api/group", {"name": "Home", "members": "Kitchen,Hallway"})
|
||||
assert api.post("/api/clean", {"slug": "Home", "what": "temp"})[0] == 200
|
||||
assert not (RUNS / "Kitchen/spirula/training").exists() and not (RUNS / "Hallway/spirula/training").exists()
|
||||
assert (RUNS / "Statue/spirula/training/x").exists()
|
||||
|
||||
|
||||
def test_dissolving_a_group_keeps_its_scenes(api):
|
||||
make_flat()
|
||||
api.post("/api/group", {"name": "Home", "members": "Kitchen,Hallway"})
|
||||
assert api.post("/api/delete", {"slug": "Home"})[0] == 200
|
||||
assert api.get("/api/groups")[1]["groups"] == []
|
||||
assert {"Kitchen", "Hallway", "Statue"} <= {s["slug"] for s in server.scenes()}
|
||||
@@ -0,0 +1,108 @@
|
||||
"""Conversions and tours as real background jobs (scripts/splat.sh, scripts/combine.sh) with a fake pipeline."""
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from conftest import RUNS, recording, server, wait_job
|
||||
import synth
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("ok_probe")
|
||||
|
||||
|
||||
def convert(name="Kitchen", existing=None, quality="fast", rec=None):
|
||||
rec = rec or recording()
|
||||
e = server.start_job({"path": str(rec), "name": name, "quality": quality, "upload": False,
|
||||
**({"existing": existing} if existing else {})})
|
||||
assert e is None, e
|
||||
return wait_job()
|
||||
|
||||
|
||||
def test_conversion_makes_a_scene():
|
||||
js = convert()
|
||||
assert js["result"] == "ok", js["tail"]
|
||||
assert js["info"]["cameras"] == 20 and js["info"]["rms_cm"] == 1.1
|
||||
s = {x["slug"]: x for x in server.scenes()}["Kitchen"]
|
||||
assert s["name"] == "Kitchen" and s["gaussians"] == 1000 and s["preview"]
|
||||
|
||||
|
||||
def test_a_second_run_of_the_same_name_asks_first():
|
||||
convert()
|
||||
assert server.start_job({"path": str(recording()), "name": "Kitchen", "quality": "fast", "upload": False}) == "exists"
|
||||
|
||||
|
||||
def test_start_over_that_fails_keeps_the_old_scene():
|
||||
convert(quality="fast")
|
||||
before = (RUNS / "Kitchen" / "scene.sog").stat().st_mtime_ns
|
||||
os.environ["FAKE_PIPELINE"] = "fail"
|
||||
js = convert(existing="fresh", quality="high")
|
||||
assert js["result"] == "failed"
|
||||
assert (RUNS / "Kitchen" / "scene.sog").stat().st_mtime_ns == before # the old result is back
|
||||
assert json.loads((RUNS / "Kitchen.ui.json").read_text())["quality"] == "fast" # with its settings
|
||||
assert not (RUNS / ".previous").exists()
|
||||
|
||||
|
||||
def test_cancelled_start_over_keeps_the_old_scene():
|
||||
convert()
|
||||
before = (RUNS / "Kitchen" / "scene.sog").stat().st_mtime_ns
|
||||
os.environ["FAKE_PIPELINE"] = "slow"
|
||||
assert server.start_job({"path": str(recording()), "name": "Kitchen", "quality": "fast", "upload": False,
|
||||
"existing": "fresh"}) is None
|
||||
time.sleep(3)
|
||||
# while it computes, the old scene stays in the list, marked
|
||||
assert {s["slug"]: s["replacing"] for s in server.scenes()}.get("Kitchen") is True
|
||||
assert server.cancel_job() is None
|
||||
js = wait_job(timeout=30)
|
||||
assert js["result"] == "cancelled"
|
||||
assert (RUNS / "Kitchen" / "scene.sog").stat().st_mtime_ns == before
|
||||
|
||||
|
||||
def test_start_over_that_works_replaces_the_scene():
|
||||
convert()
|
||||
before = (RUNS / "Kitchen" / "scene.sog").stat().st_mtime_ns
|
||||
time.sleep(0.05)
|
||||
assert convert(existing="fresh")["result"] == "ok"
|
||||
assert (RUNS / "Kitchen" / "scene.sog").stat().st_mtime_ns != before
|
||||
assert not (RUNS / ".previous").exists()
|
||||
|
||||
|
||||
def test_resume_after_an_interrupted_training():
|
||||
rec = recording()
|
||||
(RUNS / "Kitchen" / "dataset").mkdir(parents=True)
|
||||
(RUNS / "Kitchen" / "dataset" / "spirula-dataset.json").write_text('{"status": "complete"}')
|
||||
(RUNS / "Kitchen" / "spirula" / "training" / "step-18000.ckpt").mkdir(parents=True)
|
||||
(RUNS / "Kitchen" / "spirula" / "run.json").write_text('{"status": "running"}')
|
||||
(RUNS / "Kitchen.ui.json").write_text(json.dumps({"name": "Kitchen", "recording": rec.name, "quality": "fast"}))
|
||||
assert server.resume_run("Kitchen", fresh=False) is None
|
||||
js = wait_job()
|
||||
assert js["result"] == "ok", js["tail"]
|
||||
assert "Training was interrupted (last checkpoint: 18000.ckpt)" in js["tail"]
|
||||
|
||||
|
||||
def test_putting_a_tour_together():
|
||||
synth.make_room(RUNS, "Kitchen", "A", name="Kitchen", recording="2026-10-12_16-40-00_stereo.mp4")
|
||||
synth.make_room(RUNS, "Hallway", "B", name="Hallway", recording="2026-10-12_16-43-00_stereo.mp4", drift=(0.03, 0, 0))
|
||||
assert server.save_group("", "Example home", "Kitchen,Hallway") == "Example-home"
|
||||
assert server.start_combine({"slug": "Example-home", "align": "shift", "partition": "view",
|
||||
"frame_splats": "500000", "full": True}) is None
|
||||
js = wait_job()
|
||||
assert js["result"] == "ok", js["tail"]
|
||||
g = server.groups()["groups"][0]
|
||||
r = g["result"]
|
||||
assert r and r["splats"] > 10000 and r["align"] == "shift" and not r["stale"] and r["preview"]
|
||||
assert r["seams"]["Kitchen|Hallway"]["cm"] < 0.5
|
||||
tour = RUNS / "Example-home"
|
||||
assert (tour / "example-home.sog").exists() and (tour / "example-home.sceneinfo.png").exists()
|
||||
assert (tour / "full" / "example-home-full.sog").exists()
|
||||
# a scene that changes afterwards makes the tour stale
|
||||
time.sleep(0.05)
|
||||
(RUNS / "Kitchen" / "scene.sog").write_bytes(b"newer")
|
||||
assert server.groups()["groups"][0]["result"]["stale"]
|
||||
|
||||
|
||||
def test_a_tour_needs_two_scenes_and_no_other_job():
|
||||
synth.make_room(RUNS, "Kitchen", "A", name="Kitchen")
|
||||
server.save_group("", "Lonely", "Kitchen")
|
||||
assert server.start_combine({"slug": "Lonely"})["code"] == "err.combine_too_few"
|
||||
assert server.start_combine({"slug": "Lonely", "align": "magic"})["code"] == "err.bad_request"
|
||||
@@ -0,0 +1,91 @@
|
||||
"""Finished scenes: renaming, cleaning up, deleting, and runs that did not finish."""
|
||||
import json
|
||||
|
||||
from conftest import RUNS, TRASH, server
|
||||
import synth
|
||||
|
||||
|
||||
def scene_names():
|
||||
return {s["slug"]: s["name"] for s in server.scenes()}
|
||||
|
||||
|
||||
def test_finished_scenes_are_listed(api):
|
||||
synth.make_room(RUNS, "Kitchen", "A", name="Kitchen")
|
||||
status, scenes = api.get("/api/scenes")
|
||||
assert status == 200
|
||||
assert [(s["slug"], s["name"], s["ply"]) for s in scenes] == [("Kitchen", "Kitchen", True)]
|
||||
assert scenes[0]["gaussians"] > 1000
|
||||
|
||||
|
||||
def test_renaming_moves_folder_files_and_log(api):
|
||||
run = synth.make_room(RUNS, "Aufnahme-12.10.-16-40", "A", name="Aufnahme 12.10.")
|
||||
(run / "aufnahme-12.10.-16-40.sog").write_bytes(b"x") # gallery files are named after the folder
|
||||
(run / "aufnahme-12.10.-16-40.gallery.png").write_bytes(b"x")
|
||||
status, r = api.post("/api/rename", {"slug": "Aufnahme-12.10.-16-40", "new": "Büro 12.10."})
|
||||
assert status == 200, r
|
||||
new = RUNS / "Buero-12.10."
|
||||
assert new.is_dir() and not run.exists()
|
||||
assert (new / "buero-12.10..sog").exists() and (new / "buero-12.10..gallery.png").exists()
|
||||
assert (RUNS / "Buero-12.10..log").exists() and not (RUNS / "Aufnahme-12.10.-16-40.log").exists()
|
||||
meta = json.loads((RUNS / "Buero-12.10..ui.json").read_text())
|
||||
assert meta["name"] == "Büro 12.10." and meta["folder_renamed"] is True
|
||||
assert server.no_resume("Buero-12.10.") == "renamed" # the pipeline remembers the old path
|
||||
|
||||
|
||||
def test_renaming_to_a_taken_name_is_refused(api):
|
||||
synth.make_room(RUNS, "Kitchen", "A")
|
||||
synth.make_room(RUNS, "Office", "B")
|
||||
status, r = api.post("/api/rename", {"slug": "Office", "new": "Kitchen"})
|
||||
assert status == 400 and r["error"]["code"] == "err.scene_name_taken"
|
||||
assert (RUNS / "Office").is_dir()
|
||||
|
||||
|
||||
def test_cleaning_up_removes_only_what_it_says(api):
|
||||
run = synth.make_room(RUNS, "Kitchen", "A")
|
||||
for rel in ("spirula/training/step-1.ckpt", "dataset/.spirula-sfm/x", "dataset/images/cam0/0000.jpg"):
|
||||
(run / rel).parent.mkdir(parents=True, exist_ok=True)
|
||||
(run / rel).write_bytes(b"x" * 1000)
|
||||
assert api.post("/api/clean", {"slug": "Kitchen", "what": "temp"})[0] == 200
|
||||
assert not (run / "spirula/training").exists() and not (run / "dataset/.spirula-sfm").exists()
|
||||
assert (run / "dataset/images/cam0/0000.jpg").exists() # training images stay
|
||||
assert (run / "spirula/scene.ply").exists() and (run / "dataset/sparse/0/images.bin").exists()
|
||||
assert server.no_resume("Kitchen") is None
|
||||
assert api.post("/api/clean", {"slug": "Kitchen", "what": "images"})[0] == 200
|
||||
assert not (run / "dataset/images").exists()
|
||||
assert server.no_resume("Kitchen") == "images" # only a fresh start from now on
|
||||
assert api.post("/api/clean", {"slug": "Kitchen", "what": "everything"})[0] == 400
|
||||
|
||||
|
||||
def test_deleting_moves_to_the_trash(api):
|
||||
synth.make_room(RUNS, "Kitchen", "A")
|
||||
assert api.post("/api/delete", {"slug": "Kitchen"})[0] == 200
|
||||
assert "Kitchen" not in scene_names()
|
||||
assert {p.name for p in TRASH.iterdir()} >= {"Kitchen", "Kitchen.log", "Kitchen.ui.json"}
|
||||
|
||||
|
||||
def test_unfinished_runs_resume_or_start_over(api):
|
||||
# interrupted in the training: camera positions done -> it can continue
|
||||
(RUNS / "Hallway" / "dataset").mkdir(parents=True)
|
||||
(RUNS / "Hallway" / "dataset" / "spirula-dataset.json").write_text('{"status": "complete"}')
|
||||
(RUNS / "Hallway.log").write_text("Running native Spirula SfM\nstep 13801/30000 ( 46%) splats 2000000 "
|
||||
"[elapsed 00:10:00.0 | ETA 00:12:00.0] rgb_loss=0.1 ssim=0.7 psnr=25\n")
|
||||
rec = "2026-10-12_16-43-00_stereo.mp4"
|
||||
(RUNS / "Hallway.ui.json").write_text(json.dumps({"name": "Hallway", "recording": rec}))
|
||||
(server.REC / rec).write_bytes(b"x")
|
||||
# stopped while extracting images -> only from the start
|
||||
(RUNS / "Broken").mkdir()
|
||||
(RUNS / "Broken.log").write_text("Extraction progress: decoded 4% / saved 44 of 1000\n")
|
||||
(RUNS / "Broken.ui.json").write_text(json.dumps({"name": "Broken", "recording": rec}))
|
||||
(RUNS / "old-test.log").write_text("something\n") # only a log file left
|
||||
synth.make_room(RUNS, "Kitchen", "A", name="Kitchen") # finished: not in this list
|
||||
|
||||
status, runs = api.get("/api/unfinished")
|
||||
assert status == 200
|
||||
by = {r["slug"]: r for r in runs}
|
||||
assert set(by) == {"Hallway", "Broken", "old-test"}
|
||||
assert (by["Hallway"]["phase"], by["Hallway"]["pct"], by["Hallway"]["resumable"]) == (2, 46, True)
|
||||
assert (by["Broken"]["phase"], by["Broken"]["resumable"], by["Broken"]["restartable"]) == (0, False, True)
|
||||
assert by["old-test"]["phase"] is None and not by["old-test"]["restartable"]
|
||||
assert api.post("/api/resume", {"slug": "Broken"})[1]["error"]["code"] == "err.not_resumable"
|
||||
assert api.post("/api/delete", {"slug": "old-test"})[0] == 200
|
||||
assert not (RUNS / "old-test.log").exists()
|
||||
@@ -0,0 +1,95 @@
|
||||
"""Who may talk to the server: only this PC (or the home network in LAN mode), never other websites."""
|
||||
import pytest
|
||||
|
||||
from conftest import RUNS, server
|
||||
import synth
|
||||
|
||||
|
||||
def test_page_and_api_answer_locally(api):
|
||||
assert api.get("/")[0] == 200
|
||||
status, ping = api.get("/api/ping")
|
||||
assert status == 200 and ping["app"] == "splat-workshop" and ping["version"] == server.VERSION
|
||||
|
||||
|
||||
@pytest.mark.parametrize("site", ["cross-site", "same-site"])
|
||||
def test_other_websites_are_refused(api, site):
|
||||
# e.g. <img src="http://127.0.0.1:8771/api/floorplan?..."> on any website, or another port on localhost
|
||||
assert api.get("/api/scenes", {"Sec-Fetch-Site": site})[0] == 403
|
||||
assert api.post("/api/cancel", {}, {"Sec-Fetch-Site": site})[0] == 403
|
||||
|
||||
|
||||
def test_own_page_and_typed_address_work(api):
|
||||
assert api.get("/api/scenes", {"Sec-Fetch-Site": "same-origin"})[0] == 200
|
||||
assert api.get("/api/scenes", {"Sec-Fetch-Site": "none"})[0] == 200
|
||||
|
||||
|
||||
def test_a_link_may_open_the_page_but_not_the_api(api):
|
||||
nav = {"Sec-Fetch-Site": "cross-site", "Sec-Fetch-Mode": "navigate"}
|
||||
assert api.get("/", nav)[0] == 200
|
||||
assert api.get("/api/scenes", nav)[0] == 403
|
||||
|
||||
|
||||
def test_post_from_another_origin_is_refused(api):
|
||||
assert api.post("/api/cancel", {}, {"Origin": "http://evil.example"})[0] == 403
|
||||
|
||||
|
||||
def test_dns_rebinding_is_refused(api):
|
||||
# a foreign host name that resolves to 127.0.0.1
|
||||
assert api.get("/api/scenes", {"Host": "evil.example:8771"})[0] == 403
|
||||
|
||||
|
||||
@pytest.mark.parametrize("lan,client,ok", [
|
||||
(False, "127.0.0.1", True), (False, "192.168.1.20", False),
|
||||
(True, "192.168.1.20", True), (True, "10.8.0.2", True),
|
||||
(True, "100.100.1.1", False), # carrier-grade NAT / Tailscale: not a home network
|
||||
(True, "84.150.3.3", False), # the internet, e.g. through a port forwarding
|
||||
])
|
||||
def test_lan_mode_checks_the_real_client_address(monkeypatch, lan, client, ok):
|
||||
monkeypatch.setattr(server.CFG, "lan", lan)
|
||||
|
||||
class H(server.Handler):
|
||||
def __init__(self): # no socket: only what _host_ok looks at
|
||||
self.client_address = (client, 50000)
|
||||
self.headers = {"Host": "192.168.1.5:8771" if lan else "127.0.0.1:8771"}
|
||||
self.command, self.path = "GET", "/api/scenes"
|
||||
assert H()._host_ok() is ok
|
||||
|
||||
|
||||
@pytest.mark.parametrize("body,headers", [
|
||||
(b"[1, 2]", {}), # a list instead of an object
|
||||
(b'{"name": 5, "new": "x"}', {}), # a number instead of text
|
||||
(b'{"slug": {"a": 1}}', {}), # nested objects
|
||||
(b"{not json", {}),
|
||||
(b"", {"Content-Length": "-5"}),
|
||||
(b"", {"Content-Length": str(10 ** 9)}),
|
||||
])
|
||||
def test_invalid_requests_get_400_not_a_crash(api, body, headers):
|
||||
assert api.post("/api/rename", body, headers)[0] == 400
|
||||
assert api.get("/api/ping")[0] == 200 # still alive
|
||||
|
||||
|
||||
@pytest.mark.parametrize("slug", ["../runs", "..", "/etc", ".previous", "a/../../b"])
|
||||
def test_no_paths_outside_the_runs_folder(api, slug):
|
||||
synth.make_room(RUNS, "Room", "A")
|
||||
assert api.get(f"/api/preview?slug={slug}")[0] == 404
|
||||
assert api.post("/api/delete", {"slug": slug})[0] == 400
|
||||
assert api.post("/api/clean", {"slug": slug, "what": "temp"})[0] == 400
|
||||
assert (RUNS / "Room" / "scene.sog").exists()
|
||||
|
||||
|
||||
def test_actions_on_the_screen_only_from_this_pc():
|
||||
# video player, file dialog, folder, Spirula and quitting: not from another device in the LAN
|
||||
class Remote(server.Handler):
|
||||
def __init__(self):
|
||||
self.client_address = ("192.168.1.20", 50000)
|
||||
|
||||
class Local(server.Handler):
|
||||
def __init__(self):
|
||||
self.client_address = ("127.0.0.1", 50000)
|
||||
assert not Remote()._from_this_pc() and Local()._from_this_pc()
|
||||
|
||||
|
||||
def test_unknown_thumbnail_names_leave_no_cache_files(api):
|
||||
status, _ = api.get("/api/thumb?source=frame&name=whatever.mp4")
|
||||
assert status == 404
|
||||
assert not list(server.CACHE.glob("whatever*"))
|
||||
Reference in new issue
Block a user