SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat: add RELLIS dataset compatibility smoke

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-07-25 20:45:30 +03:00
parent 13c35ff446
commit 054feec0d2
19 changed files with 1633 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/datasets/__init__.py
+16
View File
@@ -38,6 +38,15 @@ from k1link.datasets.goose_qualification import (
GroundAcceptancePolicy,
qualify_goose_ground,
)
from k1link.datasets.rellis_smoke import (
RELLIS_CLASSES,
RELLIS_GROUND_POLICY_SCHEMA,
RELLIS_PREVIEW_SCHEMA,
RELLIS_SOURCE_ID,
RellisSmokeError,
build_rellis_official_smoke_preview,
rellis_native_scan_preview,
)
__all__ = [
"DATASET_GATEWAY_CATALOG_SCHEMA",
@@ -54,12 +63,18 @@ __all__ = [
"GOOSE_QUALIFICATION_PREVIEW_SCHEMA",
"GOOSE_QUALIFICATION_PROFILE_SCHEMA",
"GOOSE_QUALIFICATION_REPORT_SCHEMA",
"RELLIS_CLASSES",
"RELLIS_GROUND_POLICY_SCHEMA",
"RELLIS_PREVIEW_SCHEMA",
"RELLIS_SOURCE_ID",
"RellisSmokeError",
"GoosePatchworkProfile",
"GroundAcceptancePolicy",
"DegradationProfile",
"DEFAULT_DEGRADATIONS",
"benchmark_goose_current_ground",
"benchmark_goose_patchwork_ground",
"build_rellis_official_smoke_preview",
"configured_dataset_admission_manifest",
"configured_dataset_ground_preview",
"configured_dataset_preview",
@@ -69,5 +84,6 @@ __all__ = [
"read_dataset_ground_preview",
"read_dataset_native_scan_preview",
"read_semantic_kitti_frame",
"rellis_native_scan_preview",
"qualify_goose_ground",
]
src/k1link/datasets/cli.py
+49
View File
@@ -15,6 +15,10 @@ from k1link.datasets.goose_benchmark import (
)
from k1link.datasets.goose_qualification import qualify_goose_ground
from k1link.datasets.goose_review import build_goose_ground_review_pack
from k1link.datasets.rellis_smoke import (
RellisSmokeError,
build_rellis_official_smoke_preview,
)
app = typer.Typer(
add_completion=False,
@@ -158,5 +162,50 @@ def build_goose_ground_review_command(
typer.echo(json.dumps(manifest, ensure_ascii=False, sort_keys=True))
@app.command("build-rellis-smoke-preview")
def build_rellis_smoke_preview_command(
points: Annotated[
Path,
typer.Option("--points", exists=True, dir_okay=False, resolve_path=True),
],
labels: Annotated[
Path,
typer.Option("--labels", exists=True, dir_okay=False, resolve_path=True),
],
out: Annotated[
Path,
typer.Option("--out", dir_okay=False, resolve_path=True),
],
preview_points: Annotated[
int,
typer.Option("--preview-points", min=1, max=50_000),
] = 20_000,
) -> None:
"""Verify the pinned official RELLIS example and publish a bounded preview."""
try:
preview = build_rellis_official_smoke_preview(
points,
labels,
out,
preview_points=preview_points,
)
except RellisSmokeError as exc:
typer.echo(str(exc), err=True)
raise typer.Exit(code=2) from exc
typer.echo(
json.dumps(
{
"source_id": preview["source_id"],
"frame_id": preview["frame_id"],
"source_point_count": preview["source_point_count"],
"preview_point_count": preview["point_count"],
},
ensure_ascii=False,
sort_keys=True,
)
)
if __name__ == "__main__":
app()
src/k1link/datasets/gateway.py
+183 -7
View File
@@ -18,7 +18,7 @@ import numpy.typing as npt
from k1link.datasets.goose_profile import DEFAULT_GOOSE_PATCHWORK_PROFILE
DATASET_GATEWAY_CATALOG_SCHEMA: Final = "missioncore.dataset-gateway-catalog/v2"
DATASET_GATEWAY_CATALOG_SCHEMA: Final = "missioncore.dataset-gateway-catalog/v3"
DATASET_ADMISSION_SCHEMA: Final = "missioncore.dataset-admission/v1"
DATASET_PREVIEW_SCHEMA: Final = "missioncore.dataset-native-scan-preview/v1"
DATASET_GROUND_PREVIEW_SCHEMA: Final = "missioncore.dataset-ground-comparison-preview/v2"
@@ -177,6 +177,7 @@ def _is_worker_d_storage(root: Path | None) -> bool:
def dataset_gateway_catalog(
dataset_root: Path | None = None,
admission_manifest_path: Path | None = None,
rellis_preview_path: Path | None = None,
) -> dict[str, object]:
"""Return the path-free, read-only ingress plan and current admission state."""
@@ -196,6 +197,14 @@ def dataset_gateway_catalog(
locally_admitted = _is_worker_d_storage(root)
worker_admitted = bool(admission and admission["storage"]["admitted"])
storage_admitted = locally_admitted or worker_admitted
rellis_preview: dict[str, Any] | None = None
if rellis_preview_path is not None and rellis_preview_path.is_file():
try:
candidate = read_dataset_native_scan_preview(rellis_preview_path)
if candidate.get("source_id") == "rellis-3d/v1.1":
rellis_preview = candidate
except DatasetAdmissionError:
pass
source_status = (
str(admission["status"])
if admission is not None
@@ -212,6 +221,20 @@ def dataset_gateway_catalog(
if storage_admitted
else "configure-dataset-root-on-worker-d"
)
rellis_status = (
"smoke-ready"
if rellis_preview is not None
else "ready-for-smoke"
if storage_admitted
else "blocked-storage-policy"
)
catalog_next_action = (
"admit-rellis-ouster-semantickitti-to-worker-d"
if rellis_preview is not None and source_status == "frame-ready"
else "build-rellis-official-example-smoke"
if source_status == "frame-ready"
else next_action
)
return {
"schema_version": DATASET_GATEWAY_CATALOG_SCHEMA,
"access": "read-only",
@@ -234,11 +257,14 @@ def dataset_gateway_catalog(
"sources": [
{
"source_id": "goose-3d/v2025-08-22",
"source_kind": "goose",
"display_name": "GOOSE 3D",
"role": "primary-offroad-semantic-baseline",
"license": "CC-BY-SA-4.0",
"commercial_use": "allowed-with-share-alike",
"format": "semantickitti-xyzi-label",
"frame_semantics": "one-lidar-revolution",
"sensor": "VLS-128",
"platforms": ["MuCAR-3", "ALICE", "Spot"],
"annotations": ["semantic-point", "instance-point"],
"superclasses": [
@@ -255,10 +281,18 @@ def dataset_gateway_catalog(
"download": {
"automatic": False,
"reason": "operator-admitted-large-artifact-only",
"smoke_example_mb": None,
"primary_scan_archive_gb": 3.3,
"label_archive_gb": None,
"poses_archive_gb": None,
"training_archive_gb": 27.0,
"validation_archive_gb": 3.3,
"test_archive_gb": 3.3,
},
"statistics": {
"fragment_count": 8,
"annotated_scan_count": 961,
},
"admission": {
"status": source_status,
"native_scan": "ready-after-download",
@@ -267,7 +301,72 @@ def dataset_gateway_catalog(
"archive": admission["archive"] if admission is not None else None,
"frame": admission["frame"] if admission is not None else None,
},
}
},
{
"source_id": "rellis-3d/v1.1",
"source_kind": "rellis",
"display_name": "RELLIS-3D",
"role": "independent-offroad-cross-dataset-check",
"license": "CC-BY-NC-SA-3.0",
"commercial_use": "research-only-license-review-required",
"format": "semantickitti-xyzi-label",
"frame_semantics": "one-lidar-revolution",
"sensor": "Ouster OS1 64",
"platforms": ["Clearpath Warthog"],
"annotations": ["semantic-point"],
"superclasses": [
"ground",
"vegetation",
"structure",
"obstacle",
"vehicle",
"human",
"water",
"ignore",
],
"download": {
"automatic": False,
"reason": "operator-admitted-large-artifact-only",
"smoke_example_mb": 24.0,
"primary_scan_archive_gb": 14.0,
"label_archive_gb": 0.174,
"poses_archive_gb": 0.174,
"training_archive_gb": None,
"validation_archive_gb": None,
"test_archive_gb": None,
},
"statistics": {
"fragment_count": 5,
"annotated_scan_count": 13_556,
},
"admission": {
"status": rellis_status,
"native_scan": (
"official-example-compatible"
if rellis_preview is not None
else "requires-official-example-smoke"
),
"normalized_scan": "requires-explicit-frame-and-mounting-contract",
"rolling_local_map": "requires-poses-timing-and-map-policy",
"archive": None,
"frame": (
{
"frame_id": rellis_preview["frame_id"],
"point_count": rellis_preview["source_point_count"],
"semantic_class_count": len(rellis_preview["classes"]),
"ground_truth_ground_fraction": (
sum(rellis_preview["ground_truth_ground"])
/ rellis_preview["point_count"]
),
"preview_point_count": rellis_preview["point_count"],
"preview_sha256": None,
"preview_available": True,
}
if rellis_preview is not None
else None
),
},
},
],
"representations": [
{
@@ -330,7 +429,7 @@ def dataset_gateway_catalog(
"reason": "post-lio-map-product-cannot-be-reconstructed-as-a-native-scan",
}
],
"next_action": next_action,
"next_action": catalog_next_action,
}
@@ -417,9 +516,13 @@ def read_dataset_admission_manifest(path: Path) -> dict[str, Any]:
def read_dataset_native_scan_preview(path: Path) -> dict[str, Any]:
document = _bounded_json_object(path, MAX_PREVIEW_BYTES, "dataset preview")
identity = (document.get("schema_version"), document.get("source_id"))
if (
document.get("schema_version") != DATASET_PREVIEW_SCHEMA
or document.get("source_id") != "goose-3d/v2025-08-22"
identity
not in {
(DATASET_PREVIEW_SCHEMA, "goose-3d/v2025-08-22"),
("missioncore.dataset-native-scan-preview/v2", "rellis-3d/v1.1"),
}
or document.get("representation") != "native-scan"
or document.get("sampling") != "deterministic-even-index"
):
@@ -433,6 +536,11 @@ def read_dataset_native_scan_preview(path: Path) -> dict[str, Any]:
semantic_ids = document.get("semantic_label_ids")
semantic_rgb = document.get("semantic_rgb_0_to_255")
ground = document.get("ground_truth_ground")
evaluated = (
document.get("evaluation_mask")
if identity[1] == "rellis-3d/v1.1"
else [1] * point_count
)
if (
not isinstance(points, list)
or len(points) != point_count
@@ -444,6 +552,8 @@ def read_dataset_native_scan_preview(path: Path) -> dict[str, Any]:
or len(semantic_rgb) != point_count * 3
or not isinstance(ground, list)
or len(ground) != point_count
or not isinstance(evaluated, list)
or len(evaluated) != point_count
):
raise DatasetAdmissionError("dataset preview arrays are not point-aligned")
for point in points:
@@ -463,6 +573,7 @@ def read_dataset_native_scan_preview(path: Path) -> dict[str, Any]:
(semantic_ids, 65_535, "semantic label"),
(semantic_rgb, 255, "semantic color"),
(ground, 1, "ground mask"),
(evaluated, 1, "evaluation mask"),
):
if any(
not isinstance(value, int) or isinstance(value, bool) or not 0 <= value <= maximum
@@ -473,16 +584,81 @@ def read_dataset_native_scan_preview(path: Path) -> dict[str, Any]:
if not isinstance(classes, list) or len(classes) > 64:
raise DatasetAdmissionError("dataset preview class catalog is incompatible")
safety = _object(document.get("safety"), "safety")
if safety != {
expected_safety = {
"visualization_only": True,
"navigation_or_safety_accepted": False,
}:
}
if identity[1] == "rellis-3d/v1.1":
expected_safety["compatibility_smoke_only"] = True
if safety != expected_safety:
raise DatasetAdmissionError("dataset preview safety boundary is incompatible")
if not isinstance(document.get("frame_id"), str) or not document["frame_id"]:
raise DatasetAdmissionError("dataset preview frame id is incompatible")
if identity[1] == "rellis-3d/v1.1":
_validate_rellis_preview_metadata(document, classes)
return document
def _validate_rellis_preview_metadata(
document: dict[str, Any],
classes: list[Any],
) -> None:
coordinate_frame = _object(document.get("coordinate_frame"), "coordinate_frame")
if coordinate_frame != {
"frame_id": "sensor/lidar/os1",
"handedness": "right",
"x": "forward",
"y": "left",
"z": "up",
"transform_applied": False,
}:
raise DatasetAdmissionError("RELLIS coordinate frame is incompatible")
policy = _object(document.get("ground_policy"), "ground_policy")
if (
policy.get("schema_version") != "missioncore.rellis-ground-target-policy/v1"
or set(policy) != {"schema_version", "ground", "non_ground", "ignore"}
or any(
not isinstance(policy.get(key), list)
or any(not isinstance(value, str) or not value for value in policy[key])
for key in ("ground", "non_ground", "ignore")
)
):
raise DatasetAdmissionError("RELLIS ground target policy is incompatible")
for item in classes:
value = _object(item, "class")
if (
value.get("ground_target") not in {"ground", "non-ground", "ignore"}
or not isinstance(value.get("label_id"), int)
or not isinstance(value.get("class_name"), str)
or not isinstance(value.get("hex"), str)
or not isinstance(value.get("source_point_count"), int)
):
raise DatasetAdmissionError("RELLIS class catalog is incompatible")
evidence = _object(document.get("source_evidence"), "source_evidence")
if (
set(evidence)
!= {
"repository_url",
"repository_commit",
"label_config_sha256",
"point_sha256",
"label_sha256",
"license",
}
or evidence.get("repository_url") != "https://github.com/unmannedlab/RELLIS-3D"
or evidence.get("repository_commit")
!= "c17a118fcaed1559f03cc32cc3a91dedc557f8b8"
or evidence.get("label_config_sha256")
!= "573379a232ac561805987466c391a61fd5ad338be7fb9c4c2842f3f28067e0ad"
or evidence.get("point_sha256")
!= "ed81a9c3636d55b17d78058c72545d5d22419beecf174d50596d23ae178752af"
or evidence.get("label_sha256")
!= "9b8c65b710873e931af4ac6dfc7d3dd2298696514ab721e50300bd55ad5b634e"
or evidence.get("license") != "CC-BY-NC-SA-3.0"
):
raise DatasetAdmissionError("RELLIS source evidence is incompatible")
def read_dataset_ground_preview(path: Path) -> dict[str, Any]:
document = _bounded_json_object(path, MAX_PREVIEW_BYTES, "dataset ground preview")
if (
src/k1link/datasets/rellis_smoke.py
+275
View File
@@ -0,0 +1,275 @@
"""Pinned RELLIS-3D compatibility smoke for the shared Dataset Gateway.
The smoke check deliberately stops before algorithm qualification. It proves
that Mission Core can read one official Ouster OS1 SemanticKITTI frame, preserve
point/label alignment, apply an explicit ground-evaluation policy and publish a
bounded visualization artifact. Full RELLIS archives and ROS bags remain
worker-D-only inputs.
"""
from __future__ import annotations
import hashlib
import json
import os
import tempfile
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Final, Literal
import numpy as np
from k1link.datasets.gateway import DatasetFrameError, DatasetPointFrame, read_semantic_kitti_frame
RELLIS_SOURCE_ID: Final = "rellis-3d/v1.1"
RELLIS_PREVIEW_SCHEMA: Final = "missioncore.dataset-native-scan-preview/v2"
RELLIS_GROUND_POLICY_SCHEMA: Final = "missioncore.rellis-ground-target-policy/v1"
RELLIS_REPOSITORY_URL: Final = "https://github.com/unmannedlab/RELLIS-3D"
RELLIS_REPOSITORY_COMMIT: Final = "c17a118fcaed1559f03cc32cc3a91dedc557f8b8"
RELLIS_LABEL_CONFIG_SHA256: Final = (
"573379a232ac561805987466c391a61fd5ad338be7fb9c4c2842f3f28067e0ad"
)
RELLIS_EXAMPLE_POINTS_SHA256: Final = (
"ed81a9c3636d55b17d78058c72545d5d22419beecf174d50596d23ae178752af"
)
RELLIS_EXAMPLE_LABELS_SHA256: Final = (
"9b8c65b710873e931af4ac6dfc7d3dd2298696514ab721e50300bd55ad5b634e"
)
RELLIS_EXAMPLE_FRAME_ID: Final = "000104"
RELLIS_LICENSE: Final = "CC-BY-NC-SA-3.0"
MAX_RELLIS_PREVIEW_POINTS: Final = 50_000
GroundTarget = Literal["ground", "non-ground", "ignore"]
@dataclass(frozen=True)
class RellisClass:
label_id: int
class_name: str
bgr: tuple[int, int, int]
ground_target: GroundTarget
@property
def rgb(self) -> tuple[int, int, int]:
blue, green, red = self.bgr
return red, green, blue
@property
def hex(self) -> str:
red, green, blue = self.rgb
return f"#{red:02x}{green:02x}{blue:02x}"
def to_preview_dict(self) -> dict[str, object]:
return {
"label_id": self.label_id,
"class_name": self.class_name,
"hex": self.hex,
"ground_target": self.ground_target,
}
# IDs, names and BGR colors are pinned to the official repository config at
# RELLIS_REPOSITORY_COMMIT. The final column is Mission Core's versioned,
# algorithm-scoped ground evaluation policy; it is not an upstream claim.
RELLIS_CLASSES: Final[tuple[RellisClass, ...]] = (
RellisClass(0, "void", (0, 0, 0), "ignore"),
RellisClass(1, "dirt", (108, 64, 20), "ground"),
RellisClass(3, "grass", (0, 102, 0), "ground"),
RellisClass(4, "tree", (0, 255, 0), "non-ground"),
RellisClass(5, "pole", (0, 153, 153), "non-ground"),
RellisClass(6, "water", (0, 128, 255), "ignore"),
RellisClass(7, "sky", (0, 0, 255), "ignore"),
RellisClass(8, "vehicle", (255, 255, 0), "non-ground"),
RellisClass(9, "object", (255, 0, 127), "ignore"),
RellisClass(10, "asphalt", (64, 64, 64), "ground"),
RellisClass(12, "building", (255, 0, 0), "non-ground"),
RellisClass(15, "log", (102, 0, 0), "non-ground"),
RellisClass(17, "person", (204, 153, 255), "non-ground"),
RellisClass(18, "fence", (102, 0, 204), "non-ground"),
RellisClass(19, "bush", (255, 153, 204), "non-ground"),
RellisClass(23, "concrete", (170, 170, 170), "ground"),
RellisClass(27, "barrier", (41, 121, 255), "non-ground"),
RellisClass(31, "puddle", (134, 255, 239), "ignore"),
RellisClass(33, "mud", (99, 66, 34), "ground"),
RellisClass(34, "rubble", (110, 22, 138), "non-ground"),
)
RELLIS_CLASS_BY_ID: Final = {item.label_id: item for item in RELLIS_CLASSES}
class RellisSmokeError(RuntimeError):
"""The pinned official example cannot satisfy the RELLIS smoke contract."""
def build_rellis_official_smoke_preview(
point_path: Path,
label_path: Path,
output_path: Path,
*,
preview_points: int = 20_000,
) -> dict[str, Any]:
"""Verify the official example and publish one bounded path-free preview."""
points_digest = _sha256_file(point_path)
labels_digest = _sha256_file(label_path)
if points_digest != RELLIS_EXAMPLE_POINTS_SHA256:
raise RellisSmokeError("RELLIS official example point digest is incompatible")
if labels_digest != RELLIS_EXAMPLE_LABELS_SHA256:
raise RellisSmokeError("RELLIS official example label digest is incompatible")
try:
frame = read_semantic_kitti_frame(point_path, label_path)
except DatasetFrameError as exc:
raise RellisSmokeError("RELLIS official example violates XYZI/label alignment") from exc
preview = rellis_native_scan_preview(
frame,
frame_id=RELLIS_EXAMPLE_FRAME_ID,
maximum_points=preview_points,
source_evidence={
"repository_url": RELLIS_REPOSITORY_URL,
"repository_commit": RELLIS_REPOSITORY_COMMIT,
"label_config_sha256": RELLIS_LABEL_CONFIG_SHA256,
"point_sha256": points_digest,
"label_sha256": labels_digest,
"license": RELLIS_LICENSE,
},
)
_atomic_json(output_path, preview)
return preview
def rellis_native_scan_preview(
frame: DatasetPointFrame,
*,
frame_id: str,
maximum_points: int,
source_evidence: dict[str, str],
) -> dict[str, Any]:
"""Build the common viewer artifact from a validated RELLIS native frame."""
if not frame_id or not 1 <= maximum_points <= MAX_RELLIS_PREVIEW_POINTS:
raise RellisSmokeError("RELLIS preview bounds are incompatible")
unknown = sorted(
int(value)
for value in np.unique(frame.semantic_labels)
if int(value) not in RELLIS_CLASS_BY_ID
)
if unknown:
raise RellisSmokeError("RELLIS frame contains labels absent from the pinned ontology")
required_evidence = {
"repository_url",
"repository_commit",
"label_config_sha256",
"point_sha256",
"label_sha256",
"license",
}
if set(source_evidence) != required_evidence or any(
not isinstance(value, str) or not value for value in source_evidence.values()
):
raise RellisSmokeError("RELLIS source evidence is incomplete")
sample_count = min(frame.point_count, maximum_points)
indices = np.linspace(0, frame.point_count - 1, sample_count, dtype=np.int64)
points = frame.points_xyz_m[indices]
remission = frame.remission[indices]
semantic = frame.semantic_labels[indices]
remission_min = float(np.min(remission))
remission_max = float(np.max(remission))
remission_span = remission_max - remission_min
if remission_span <= 0:
remission_u8 = np.zeros(sample_count, dtype=np.uint8)
else:
remission_u8 = np.rint(
(remission - remission_min) / remission_span * 255.0
).astype(np.uint8)
ground = np.zeros(sample_count, dtype=np.uint8)
evaluated = np.zeros(sample_count, dtype=np.uint8)
colors = np.zeros((sample_count, 3), dtype=np.uint8)
for label_id in np.unique(semantic):
item = RELLIS_CLASS_BY_ID[int(label_id)]
mask = semantic == label_id
colors[mask] = item.rgb
if item.ground_target != "ignore":
evaluated[mask] = 1
if item.ground_target == "ground":
ground[mask] = 1
present = Counter(int(value) for value in frame.semantic_labels)
present_classes = [
{
**RELLIS_CLASS_BY_ID[label_id].to_preview_dict(),
"source_point_count": count,
}
for label_id, count in sorted(present.items())
]
return {
"schema_version": RELLIS_PREVIEW_SCHEMA,
"source_id": RELLIS_SOURCE_ID,
"frame_id": frame_id,
"representation": "native-scan",
"sampling": "deterministic-even-index",
"source_point_count": frame.point_count,
"point_count": sample_count,
"points_xyz_m": points.tolist(),
"remission_0_to_255": remission_u8.tolist(),
"semantic_label_ids": semantic.astype(np.uint16).tolist(),
"semantic_rgb_0_to_255": colors.reshape(-1).tolist(),
"ground_truth_ground": ground.tolist(),
"evaluation_mask": evaluated.tolist(),
"classes": present_classes,
"coordinate_frame": {
"frame_id": "sensor/lidar/os1",
"handedness": "right",
"x": "forward",
"y": "left",
"z": "up",
"transform_applied": False,
},
"ground_policy": {
"schema_version": RELLIS_GROUND_POLICY_SCHEMA,
"ground": [
item.class_name for item in RELLIS_CLASSES if item.ground_target == "ground"
],
"non_ground": [
item.class_name
for item in RELLIS_CLASSES
if item.ground_target == "non-ground"
],
"ignore": [
item.class_name for item in RELLIS_CLASSES if item.ground_target == "ignore"
],
},
"source_evidence": source_evidence,
"safety": {
"visualization_only": True,
"compatibility_smoke_only": True,
"navigation_or_safety_accepted": False,
},
}
def _sha256_file(path: Path) -> str:
digest = hashlib.sha256()
try:
with path.open("rb") as source:
for chunk in iter(lambda: source.read(1024**2), b""):
digest.update(chunk)
except OSError as exc:
raise RellisSmokeError("RELLIS official example is unavailable") from exc
return digest.hexdigest()
def _atomic_json(path: Path, document: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(
mode="w",
dir=path.parent,
encoding="utf-8",
delete=False,
) as temporary:
temporary_path = Path(temporary.name)
json.dump(document, temporary, ensure_ascii=False, separators=(",", ":"))
temporary.flush()
os.fsync(temporary.fileno())
os.replace(temporary_path, path)