239 lines
8.1 KiB
Python
239 lines
8.1 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from fastapi import APIRouter
|
|
from fastapi.routing import APIRoute
|
|
|
|
from k1link.compute import (
|
|
E10_LIDAR_PACK_SCHEMA,
|
|
E10LidarFieldSource,
|
|
FieldReviewWindowSpec,
|
|
GroundBenchmarkProfile,
|
|
GroundSegmentation,
|
|
LidarFieldReviewV1,
|
|
LidarGroundError,
|
|
build_lidar_field_review,
|
|
)
|
|
from k1link.web.lidar_api import build_lidar_router
|
|
|
|
|
|
class _Candidate:
|
|
@property
|
|
def identity(self) -> dict[str, object]:
|
|
return {
|
|
"provider_id": "test-patchwork/v1",
|
|
"source_commit": "c" * 40,
|
|
}
|
|
|
|
def segment(self, xyzi: np.ndarray) -> GroundSegmentation:
|
|
ground = xyzi[:, 0] <= 0.25
|
|
assigned = np.ones(xyzi.shape[0], dtype=np.bool_)
|
|
return GroundSegmentation(ground, assigned, 0.25)
|
|
|
|
|
|
def _canonical_json(value: object) -> bytes:
|
|
return json.dumps(
|
|
value,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
allow_nan=False,
|
|
).encode()
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def _source_pack(root: Path) -> Path:
|
|
base_points = np.asarray(
|
|
[
|
|
[-1.0, -1.0, 0.00],
|
|
[-0.5, -1.0, 0.02],
|
|
[0.0, -1.0, 0.01],
|
|
[0.5, -1.0, 0.03],
|
|
[1.0, -1.0, 0.00],
|
|
[-1.0, 0.0, 0.01],
|
|
[-0.5, 0.0, 0.02],
|
|
[0.0, 0.0, 0.04],
|
|
[0.5, 0.0, 0.35],
|
|
[1.0, 0.0, 0.70],
|
|
],
|
|
dtype=np.float32,
|
|
)
|
|
points = np.concatenate(
|
|
[base_points + np.asarray([index * 0.1, index, 0], dtype=np.float32) for index in range(4)]
|
|
).astype("<f4")
|
|
arrays = {
|
|
"frame_indices": np.arange(4, dtype="<i8"),
|
|
"source_frame_indices": np.asarray([100, 110, 120, 130], dtype="<i8"),
|
|
"session_seconds": np.asarray([1.0, 2.0, 3.0, 4.0], dtype="<f8"),
|
|
"sample_available": np.ones(4, dtype="?"),
|
|
"cloud_offsets": np.asarray([0, 10, 20, 30, 40], dtype="<i8"),
|
|
"cloud_points_map": points,
|
|
"pose_positions_map": np.zeros((4, 3), dtype="<f8"),
|
|
"pose_quaternions_map_from_lidar": np.tile(
|
|
np.asarray([0.0, 0.0, 0.0, 1.0], dtype="<f8"),
|
|
(4, 1),
|
|
),
|
|
"lidar_camera_delta_ms": np.zeros(4, dtype="<f8"),
|
|
"pose_point_delta_ms": np.zeros(4, dtype="<f8"),
|
|
"intrinsic_fx_fy_cx_cy": np.asarray(
|
|
[100.0, 100.0, 50.0, 50.0],
|
|
dtype="<f8",
|
|
),
|
|
"distortion_kb4": np.zeros(4, dtype="<f8"),
|
|
"t_camera_from_lidar": np.eye(4, dtype="<f8"),
|
|
}
|
|
identity = {
|
|
"schema_version": E10_LIDAR_PACK_SCHEMA,
|
|
"session_id": "synthetic-ravnoves00",
|
|
"frame_count": 4,
|
|
"available_lidar_frames": 4,
|
|
"point_count": 40,
|
|
"timeline_start_seconds": 1.0,
|
|
"timeline_end_seconds": 4.0,
|
|
}
|
|
identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
|
|
pack = root / f"e10-lidar-pack-{identity_sha256}"
|
|
pack.mkdir(parents=True)
|
|
arrays_path = pack / "lidar-pack.npz"
|
|
np.savez_compressed(arrays_path, **arrays) # type: ignore[arg-type]
|
|
manifest = {
|
|
"schema_version": E10_LIDAR_PACK_SCHEMA,
|
|
"pack_id": pack.name,
|
|
"identity_sha256": identity_sha256,
|
|
"identity": identity,
|
|
"artifact": {
|
|
"path": arrays_path.name,
|
|
"media_type": "application/x-npz",
|
|
"byte_length": arrays_path.stat().st_size,
|
|
"sha256": _sha256(arrays_path),
|
|
},
|
|
}
|
|
(pack / "manifest.json").write_bytes(json.dumps(manifest, sort_keys=True).encode())
|
|
return pack
|
|
|
|
|
|
def _endpoint(router: APIRouter, path: str) -> object:
|
|
for route in router.routes:
|
|
if (
|
|
isinstance(route, APIRoute)
|
|
and route.path == path
|
|
and route.methods is not None
|
|
and "GET" in route.methods
|
|
):
|
|
return route.endpoint
|
|
raise AssertionError(f"GET {path} route is missing")
|
|
|
|
|
|
def test_field_review_is_accumulated_path_free_visual_evidence(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source = E10LidarFieldSource(_source_pack(tmp_path / "source"))
|
|
windows = (
|
|
FieldReviewWindowSpec("street-a", "Улица A", 0.5, 2.1, 100),
|
|
FieldReviewWindowSpec("street-b", "Улица B", 2.5, 4.1, 120),
|
|
)
|
|
previews = {}
|
|
for window in windows:
|
|
preview = tmp_path / f"{window.key}.jpg"
|
|
preview.write_bytes(b"\xff\xd8synthetic-jpeg\xff\xd9")
|
|
previews[window.key] = preview
|
|
profile = GroundBenchmarkProfile(
|
|
profile_id="synthetic-ravnoves00-field-review/v1",
|
|
patchwork_sensor_height_proxy_m=1.27,
|
|
patchwork_map_vertical_origin_offset_m=1.27,
|
|
patchwork_height_evidence="operator-estimated",
|
|
)
|
|
try:
|
|
output = build_lidar_field_review(
|
|
source,
|
|
tmp_path / "reviews",
|
|
patchwork=_Candidate(),
|
|
profile=profile,
|
|
preview_paths=previews,
|
|
windows=windows,
|
|
default_window_index=1,
|
|
maximum_display_points=12,
|
|
)
|
|
finally:
|
|
source.close()
|
|
|
|
review = LidarFieldReviewV1(output)
|
|
try:
|
|
detail = review.window_detail(1)
|
|
assert review.report["status"] == "diagnostic-only"
|
|
assert review.report["source"]["intensity_available"] is False
|
|
assert review.report["selection"]["default_window_index"] == 1
|
|
assert len(review.report["windows"]) == 2
|
|
assert review.report["windows"][0]["source_lidar_samples"] == 2
|
|
assert detail["point_count"] == 12
|
|
intensity = detail["intensity"]
|
|
authority = detail["authority"]
|
|
assert isinstance(intensity, dict)
|
|
assert isinstance(authority, dict)
|
|
assert intensity["available"] is False
|
|
assert detail["ground_truth"] is False
|
|
assert authority["commands_enabled"] is False
|
|
assert str(tmp_path) not in repr(detail)
|
|
finally:
|
|
review.close()
|
|
|
|
router = build_lidar_router(
|
|
root_provider=lambda: None,
|
|
ground_root_provider=lambda: None,
|
|
field_review_root_provider=lambda: output.parent,
|
|
)
|
|
catalog_route = _endpoint(router, "/api/v1/lidar/field-reviews")
|
|
window_route = _endpoint(
|
|
router,
|
|
"/api/v1/lidar/field-reviews/{review_id}/windows/{window_index}",
|
|
)
|
|
preview_route = _endpoint(
|
|
router,
|
|
"/api/v1/lidar/field-reviews/{review_id}/windows/{window_index}/preview",
|
|
)
|
|
catalog = catalog_route(limit=10) # type: ignore[operator]
|
|
window = window_route(review_id=output.name, window_index=0) # type: ignore[operator]
|
|
preview = preview_route(review_id=output.name, window_index=0) # type: ignore[operator]
|
|
|
|
assert catalog["valid_total"] == 1
|
|
assert catalog["items"][0]["display_name"]
|
|
assert window["preview_url"].endswith("/windows/0/preview")
|
|
assert preview.media_type == "image/jpeg"
|
|
assert preview.body.startswith(b"\xff\xd8")
|
|
assert str(tmp_path) not in repr({"catalog": catalog, "window": window})
|
|
|
|
|
|
def test_field_review_reader_rejects_tampered_identity(tmp_path: Path) -> None:
|
|
source = E10LidarFieldSource(_source_pack(tmp_path / "source"))
|
|
window = FieldReviewWindowSpec("street-a", "Улица A", 0.5, 4.1, 100)
|
|
preview = tmp_path / "street-a.jpg"
|
|
preview.write_bytes(b"\xff\xd8synthetic-jpeg\xff\xd9")
|
|
try:
|
|
output = build_lidar_field_review(
|
|
source,
|
|
tmp_path / "reviews",
|
|
patchwork=_Candidate(),
|
|
profile=GroundBenchmarkProfile(),
|
|
preview_paths={window.key: preview},
|
|
windows=(window,),
|
|
default_window_index=0,
|
|
maximum_display_points=12,
|
|
)
|
|
finally:
|
|
source.close()
|
|
|
|
manifest_path = output / "manifest.json"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
manifest["identity"]["display_name"] = "tampered"
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
with pytest.raises(LidarGroundError, match="identity"):
|
|
LidarFieldReviewV1(output)
|