Первоначальная настройка
This commit is contained in:
@@ -0,0 +1,49 @@
|
||||
import json
|
||||
|
||||
import numpy as np
|
||||
|
||||
from iiwa_genesis.camera import D455, Frame, calibration, optical_pose, optical_rays, process_depth
|
||||
from iiwa_genesis.config import ROOT, load_config
|
||||
|
||||
|
||||
def camera_config():
|
||||
return load_config(ROOT / "configs/scene.yaml")["cameras"]["overhead"]
|
||||
|
||||
|
||||
def test_downward_pose():
|
||||
cfg = camera_config()
|
||||
pose = optical_pose(cfg["position"], cfg["lookat"], cfg["up"])
|
||||
np.testing.assert_allclose(pose[:3, :3].T @ pose[:3, :3], np.eye(3))
|
||||
np.testing.assert_allclose(pose[:3, 2], [0, 0, -1])
|
||||
assert np.linalg.det(pose[:3, :3]) > 0.999
|
||||
|
||||
|
||||
def test_plane_range_converts_to_constant_z():
|
||||
cfg = camera_config()
|
||||
distances = 2 * np.linalg.norm(optical_rays(cfg), axis=-1)
|
||||
depth, valid = process_depth(distances, cfg, np.random.default_rng(1))
|
||||
np.testing.assert_allclose(depth, 2)
|
||||
assert valid.all()
|
||||
distances[0, :3] = [np.inf, 0, 100]
|
||||
depth, valid = process_depth(distances, cfg, np.random.default_rng(1))
|
||||
assert not valid[0, :3].any()
|
||||
assert not depth[0, :3].any()
|
||||
|
||||
|
||||
def test_seeded_noise_and_serialization(tmp_path):
|
||||
cfg = camera_config()
|
||||
cfg["noise"]["enabled"] = True
|
||||
distances = 2 * np.linalg.norm(optical_rays(cfg), axis=-1)
|
||||
depth, valid = process_depth(distances, cfg, np.random.default_rng(7))
|
||||
second, _ = process_depth(distances, cfg, np.random.default_rng(7))
|
||||
np.testing.assert_array_equal(depth, second)
|
||||
assert np.std(depth) > 0
|
||||
camera = object.__new__(D455)
|
||||
camera.name, camera.cfg = "overhead", cfg
|
||||
frame = Frame(np.zeros((480, 640, 3), dtype=np.uint8), depth, valid,
|
||||
{"intrinsics": calibration(cfg).tolist(), "world_from_optical": np.eye(4).tolist()})
|
||||
camera.save(frame, tmp_path, 0, pointcloud=True)
|
||||
data = np.load(tmp_path / "overhead_000000.npz")
|
||||
assert data["depth"].shape == (480, 640)
|
||||
assert data["xyz_optical"].shape == (480, 640, 3)
|
||||
assert json.loads((tmp_path / "overhead_000000.json").read_text()) == frame.metadata
|
||||
@@ -0,0 +1,76 @@
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
from iiwa_genesis.config import ROOT, DESCRIPTION, load_config
|
||||
|
||||
|
||||
def config_file(tmp_path, mode="without_gripper"):
|
||||
data = yaml.safe_load((ROOT / "configs/scene.yaml").read_text())
|
||||
data["scene"]["mode"] = mode
|
||||
file = tmp_path / "scene.yaml"
|
||||
file.write_text(yaml.safe_dump(data))
|
||||
return file
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode,count", [("without_gripper", 1), ("with_gripper", 2)])
|
||||
def test_mode_selects_urdf_and_cameras(tmp_path, mode, count):
|
||||
cfg = load_config(config_file(tmp_path, mode))
|
||||
assert len(cfg["cameras"]) == count
|
||||
assert cfg["robot"]["urdf"].name == f"iiwa_{mode}.urdf"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["without_gripper", "with_gripper"])
|
||||
def test_urdf_tree_and_assets(mode):
|
||||
path = DESCRIPTION / f"urdf/iiwa_{mode}.urdf"
|
||||
root = ET.parse(path).getroot()
|
||||
links = {l.get("name") for l in root.findall("link")}
|
||||
joints = root.findall("joint")
|
||||
assert sum(j.get("type") == "revolute" for j in joints) == 7
|
||||
children = [j.find("child").get("link") for j in joints]
|
||||
assert len(children) == len(set(children))
|
||||
assert links - set(children) == {"world"}
|
||||
assert all(j.find("parent").get("link") in links for j in joints)
|
||||
assert ("camera_optical_frame" in links) == (mode == "with_gripper")
|
||||
for mesh in root.iter("mesh"):
|
||||
assert (path.parent / mesh.get("filename")).is_file()
|
||||
text = path.read_text()
|
||||
for prohibited in ("package://", "xacro", "ros2_control", "<gazebo", "<webots", "$(find"):
|
||||
assert prohibited not in text
|
||||
|
||||
|
||||
def test_tool_transforms_are_preserved():
|
||||
original = ET.parse(ROOT / "iiwa_description/urdf/tools/patron.xacro").getroot()
|
||||
exported = ET.parse(DESCRIPTION / "urdf/iiwa_with_gripper.urdf").getroot()
|
||||
for joint in original.findall("joint"):
|
||||
actual = exported.find(f"joint[@name='{joint.get('name')}']")
|
||||
assert actual is not None
|
||||
assert actual.attrib == joint.attrib
|
||||
for tag in ("origin", "parent", "child"):
|
||||
assert actual.find(tag).attrib == joint.find(tag).attrib
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", ["mode", "joints", "up", "fps", "range"])
|
||||
def test_invalid_config(tmp_path, bad):
|
||||
file = config_file(tmp_path)
|
||||
data = yaml.safe_load(file.read_text())
|
||||
if bad == "mode":
|
||||
data["scene"]["mode"] = "invalid"
|
||||
elif bad == "joints":
|
||||
data["robot"]["joints"]["joint1"] = 100
|
||||
elif bad == "up":
|
||||
data["cameras"]["overhead"]["up"] = [0, 0, 1]
|
||||
elif bad == "fps":
|
||||
data["cameras"]["overhead"]["fps"] = 0
|
||||
else:
|
||||
data["cameras"]["overhead"]["min_range"] = 10
|
||||
file.write_text(yaml.safe_dump(data))
|
||||
with pytest.raises(ValueError):
|
||||
load_config(file)
|
||||
|
||||
|
||||
def test_table_mesh_paths():
|
||||
path = DESCRIPTION / "table/table.urdf"
|
||||
for mesh in ET.parse(path).iter("mesh"):
|
||||
assert (path.parent / mesh.get("filename")).is_file()
|
||||
@@ -0,0 +1,43 @@
|
||||
"""Explicit opt-in: compiles Genesis kernels and needs an EGL-capable OpenGL stack."""
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
from iiwa_genesis.config import ROOT
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.environ.get("IIWA_GENESIS_SMOKE") != "1", reason="set IIWA_GENESIS_SMOKE=1 to run real CPU/EGL tests")
|
||||
@pytest.mark.parametrize("mode,names", [("without_gripper", ["overhead"]), ("with_gripper", ["overhead", "tool"])])
|
||||
def test_cpu_rgbd_scene(tmp_path, mode, names):
|
||||
cfg = yaml.safe_load((ROOT / "configs/scene.yaml").read_text())
|
||||
cfg["scene"]["mode"] = mode
|
||||
cfg["output"]["directory"] = str(tmp_path / "frames")
|
||||
path = tmp_path / "scene.yaml"
|
||||
path.write_text(yaml.safe_dump(cfg))
|
||||
env = os.environ.copy()
|
||||
env.pop("PYTHONPATH", None)
|
||||
result = subprocess.run([sys.executable, str(ROOT / "run_scene.py"), "--config", str(path),
|
||||
"--headless", "--backend", "cpu", "--steps", "4", "--save-frames"],
|
||||
env=env, capture_output=True, text=True, timeout=600)
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
captures = list((tmp_path / "frames").glob("*/*.npz"))
|
||||
assert len(captures) == len(names)
|
||||
for name in names:
|
||||
capture = next(p for p in captures if p.stem.startswith(name))
|
||||
data = np.load(capture)
|
||||
assert np.array(Image.open(capture.with_suffix(".png"))).shape == (480, 640, 3)
|
||||
assert data["depth"].shape == (480, 640)
|
||||
valid = data["valid"]
|
||||
assert np.isfinite(data["depth"]).all()
|
||||
assert (data["depth"][~valid] == 0).all()
|
||||
assert (data["depth"][valid] >= .01).all() and (data["depth"][valid] <= 6).all()
|
||||
meta = json.loads(capture.with_suffix(".json").read_text())
|
||||
np.testing.assert_allclose(np.array(meta["optical_from_world"]) @ meta["world_from_optical"], np.eye(4), atol=1e-6)
|
||||
if name == "overhead":
|
||||
assert valid.any()
|
||||
@@ -0,0 +1,18 @@
|
||||
import pytest
|
||||
|
||||
from iiwa_genesis.runtime import plain_urdf_source
|
||||
|
||||
|
||||
def test_adapter_disables_only_xacro_path():
|
||||
source = "import xacro\ndef load_xacro(path, mappings):\n return xacro.process_file(path)\ndef load_plain():\n return 7\n"
|
||||
namespace = {}
|
||||
exec(plain_urdf_source(source, "test.py"), namespace)
|
||||
assert namespace["load_plain"]() == 7
|
||||
assert "xacro" not in namespace
|
||||
with pytest.raises(RuntimeError, match="disabled"):
|
||||
namespace["load_xacro"]("foo.xacro", {})
|
||||
|
||||
|
||||
def test_adapter_rejects_changed_upstream():
|
||||
with pytest.raises(RuntimeError, match="changed"):
|
||||
plain_urdf_source("def load_plain(): pass", "test.py")
|
||||
Reference in New Issue
Block a user