mirror of
https://github.com/thunderheronight/splat-workshop.git
synced 2026-10-18 10:01:29 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a8a6aa5188 | ||
|
|
49124eacc8 | ||
|
|
4f339b2150 |
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
|
||||
@@ -2,6 +2,10 @@
|
||||
|
||||
**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/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).
|
||||
|
||||
[English](README.md)
|
||||
|
||||

|
||||
@@ -26,6 +30,9 @@ Arcturus Vision stellt die Rekonstruktions-Skripte und einen Spirula-Studio-Fork
|
||||
- 🧹 **Speicherplatz, Aufräumen, Löschen** pro Szene und eine Liste **abgebrochener Läufe** zum Fortsetzen oder Entfernen.
|
||||
- 🛡️ **Sicherer**: *Neu rechnen* behält die alte Szene, bis die neue fertig ist; der PC geht während einer Umwandlung nicht mehr in den Standby.
|
||||
|
||||
<img src="docs/groups.png" width="800" alt="Eine Gruppe „Example home“ mit sechs Räumen: der daraus zusammengesetzte Rundgang, sein Grundriss als Vorschau, wie gut die Räume aneinanderpassen, darunter die Räume">
|
||||
<br><sub>Beispieldaten (englische Oberfläche) – Raumnamen und Vorschaubilder sind Platzhalter, die Rundgang-Vorschau ist die synthetische Beispielwohnung.</sub>
|
||||
|
||||
Alle Änderungen: [CHANGELOG.md](CHANGELOG.md) (englisch)
|
||||
|
||||
## Funktionen
|
||||
@@ -36,6 +43,8 @@ Alle Änderungen: [CHANGELOG.md](CHANGELOG.md) (englisch)
|
||||
- **Abbrechen & Fortsetzen** ab dem letzten Zwischenstand.
|
||||
- **Aufnahmeprüfung ohne KI**: reine Messwerte je Sekunde, farbige Zeitleiste und konkrete Hinweise („0:45–0:59 wenig Struktur“).
|
||||
- **Gruppen und Rundgänge**: Szenen gruppieren (z. B. alle Räume der Wohnung) und zu einer Szene zusammensetzen – eine volle Fassung für Gaussian XR am PC und eine kleinere für die Frame. Aufnahmen aus einem Durchgang teilen das Tracking-Koordinatensystem der Frame; die Werkstatt richtet die Räume aneinander aus (Korrekturen A–D zur Wahl), legt die Grenzen zwischen den Aufnahmen in die Türen und zeigt, wie gut die Räume aneinanderpassen. Vor dem Start schätzt sie den Speicherbedarf.
|
||||
<br><img src="docs/combine.png" width="360" alt="Dialog „Zusammensetzen“: Korrekturen A bis D, Grenzen zwischen den Aufnahmen, Größe der Frame-Fassung, volle Fassung, Speicherschätzung">
|
||||
<br><sub>Beispieldaten (englische Oberfläche) – Raumnamen und Vorschaubilder sind Platzhalter.</sub>
|
||||
- **Grundrisse** *(experimentell)* einzelner Szenen oder einer ganzen Gruppe – Aufnahmen aus einem Durchgang fügen sich von selbst zum Grundriss der ganzen Wohnung. Zwei Stile: *Hübsch* (Draufsicht in Farbe, Wände, gelaufener Weg) und *Klassisch* (leuchtende Punktdichte), dazu ein einfaches Messwerkzeug (Punkte anklicken, Abstände ablesen – etwa ±5 cm genau).
|
||||
- **Fertige Szenen**: später auf die Frame übertragen, erneut übertragen, umbenennen (Ordner und Namensschild ziehen mit), löschen (in den Papierkorb), Ordner oder Spirula Studio öffnen.
|
||||
- **Abgebrochene Läufe** werden mit ihrem Stand angezeigt: fortsetzen, neu starten oder löschen.
|
||||
|
||||
@@ -2,6 +2,10 @@
|
||||
|
||||
**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/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).
|
||||
|
||||
[Deutsch](README.de.md)
|
||||
|
||||

|
||||
@@ -28,6 +32,9 @@ Arcturus Vision publishes the reconstruction scripts and a Spirula Studio fork,
|
||||
- 🧹 **Disk usage, clean-up, delete** per scene, and a list of **cancelled runs** to resume or remove.
|
||||
- 🛡️ **Safer**: *Start over* keeps the old scene until the new one is done; the PC no longer goes to sleep mid-conversion.
|
||||
|
||||
<img src="docs/groups.png" width="800" alt="A group “Example home” with six rooms: the tour put together from them, its floor plan as preview, how well the rooms meet, and the rooms below">
|
||||
<br><sub>Example data – room names and preview images are placeholders; the tour preview is the synthetic example home.</sub>
|
||||
|
||||
All changes: [CHANGELOG.md](CHANGELOG.md)
|
||||
|
||||
## Features
|
||||
@@ -38,6 +45,8 @@ All changes: [CHANGELOG.md](CHANGELOG.md)
|
||||
- **Cancel & resume**: a cancelled conversion continues from its last saved state.
|
||||
- **Recording check** without AI: plain per-second measurements with a colour-coded timeline and concrete hints (“0:45–0:59 little structure”). See [What the check measures](#what-the-check-measures).
|
||||
- **Groups and tours**: group scenes (e.g. all rooms of your home) and put them together into one scene – a full version for Gaussian XR on the PC and a smaller one for the Frame. Recordings from one session share the Frame's tracking frame; Splat Workshop fits the rooms to each other (corrections A–D to choose), lets the borders between recordings run through doors and shows how well the rooms meet. Before starting it estimates the memory needed.
|
||||
<br><img src="docs/combine.png" width="360" alt="Dialog “Put together”: corrections A to D, borders between the recordings, size of the Frame version, full version, memory estimate">
|
||||
<br><sub>Example data – room names and preview images are placeholders.</sub>
|
||||
- **Floor plans** *(experimental)* of single scenes or a whole group – recordings from one session line up by themselves into a plan of your whole home. Two styles: *Pretty* (colour top view, walls, the path you walked) and *Classic* (glowing point density), plus a simple measuring tool (click points, get distances – about ±5 cm).
|
||||
<br><img src="docs/floorplan.png" width="640" alt="Combined floor plan of an example home in the classic style, with the measuring tool">
|
||||
<br><sub>Example home – synthetic data, drawn with the same renderer. Combining rooms relies on the Frame keeping its tracking frame between recordings; how well that holds for a whole home is still being tested.</sub>
|
||||
@@ -156,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.
|
||||
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 103 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 267 KiB |
@@ -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