"""Apply a human-labelled point-cloud semantic model to a new RGB point cloud.
|
|
This script deliberately accepts only explicit model/input/output paths. The
|
workbench server supplies those paths from fixed directories; it never forwards
|
browser paths or commands to this process.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import hashlib
|
import json
|
import sys
|
import time
|
from datetime import UTC, datetime
|
from pathlib import Path
|
from typing import Any
|
|
import laspy
|
import numpy as np
|
import open3d as o3d
|
import torch
|
|
from train_pointcloud_semantic_model import CLASS_SCHEMA, PointWiseNet
|
|
|
PREVIEW_POINT_LIMIT = 400_000
|
|
|
def sha256(path: Path) -> str:
|
digest = hashlib.sha256()
|
with path.open("rb") as stream:
|
for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""):
|
digest.update(chunk)
|
return digest.hexdigest()
|
|
|
def source_has_rgb(path: Path) -> bool:
|
suffix = path.suffix.lower()
|
if suffix in {".las", ".laz"}:
|
header = laspy.read(path).header
|
return all(name in header.point_format.dimension_names for name in ("red", "green", "blue"))
|
if suffix == ".ply":
|
with path.open("rb") as stream:
|
header = stream.read(64 * 1024).decode("latin-1", errors="ignore").lower()
|
return "end_header" in header and all(f" {name}" in header for name in ("red", "green", "blue"))
|
if suffix == ".pcd":
|
with path.open("rb") as stream:
|
header = stream.read(64 * 1024).decode("latin-1", errors="ignore").lower()
|
fields = next((line for line in header.splitlines() if line.startswith("fields ")), "")
|
return " rgb" in f" {fields}" or " rgba" in f" {fields}"
|
return False
|
|
|
def load_cloud(path: Path) -> tuple[np.ndarray, np.ndarray, Any | None]:
|
if path.suffix.lower() in {".las", ".laz"}:
|
las = laspy.read(path)
|
dimensions = set(las.point_format.dimension_names)
|
if not {"red", "green", "blue"}.issubset(dimensions):
|
raise ValueError("LAS/LAZ input has no RGB dimensions; this model requires observed RGB.")
|
xyz = np.column_stack((las.x, las.y, las.z)).astype(np.float32)
|
raw_rgb = np.column_stack((las.red, las.green, las.blue)).astype(np.float32)
|
maximum = float(raw_rgb.max())
|
if maximum <= 0:
|
raise ValueError("LAS/LAZ input RGB values are all zero; observed RGB is required.")
|
return xyz, np.clip(raw_rgb / max(maximum, 1.0), 0.0, 1.0), las
|
if not source_has_rgb(path):
|
raise ValueError("Input has no readable RGB fields; XYZ-only point clouds cannot use this trained model.")
|
cloud = o3d.io.read_point_cloud(str(path))
|
xyz = np.asarray(cloud.points, dtype=np.float32)
|
rgb = np.asarray(cloud.colors, dtype=np.float32)
|
if len(xyz) < 1 or xyz.shape != rgb.shape:
|
raise ValueError("Input point cloud has no readable XYZ/RGB points.")
|
return xyz, np.clip(rgb, 0.0, 1.0), None
|
|
|
def preview_indices(predictions: np.ndarray, limit: int) -> np.ndarray:
|
"""Deterministically retain every small predicted class before filling the budget."""
|
count = len(predictions)
|
if count <= limit:
|
return np.arange(count, dtype=np.int64)
|
rng = np.random.default_rng(42)
|
selected: list[np.ndarray] = []
|
remaining = limit
|
groups = sorted((np.flatnonzero(predictions == code) for code in np.unique(predictions)), key=len)
|
for group in groups:
|
keep = min(len(group), max(1, min(50_000, remaining // max(1, len(groups) - len(selected)))))
|
selected.append(group if keep == len(group) else np.sort(rng.choice(group, size=keep, replace=False)))
|
remaining -= keep
|
chosen = np.concatenate(selected)
|
if len(chosen) < limit:
|
mask = np.ones(count, dtype=bool)
|
mask[chosen] = False
|
fill = rng.choice(np.flatnonzero(mask), size=limit - len(chosen), replace=False)
|
chosen = np.concatenate((chosen, fill))
|
if len(chosen) > limit:
|
chosen = rng.choice(chosen, size=limit, replace=False)
|
return np.sort(chosen.astype(np.int64))
|
|
|
def load_model(path: Path, device: torch.device) -> tuple[PointWiseNet, list[int]]:
|
payload = torch.load(path, map_location=device, weights_only=False)
|
if not isinstance(payload, dict) or payload.get("schema_version") != 1:
|
raise ValueError("Unsupported model schema.")
|
codes = payload.get("class_codes")
|
if not isinstance(codes, list) or len(codes) < 2 or any(not isinstance(code, int) or code not in CLASS_SCHEMA for code in codes):
|
raise ValueError("Model class schema is invalid.")
|
model = PointWiseNet(len(codes)).to(device)
|
try:
|
model.load_state_dict(payload["state_dict"], strict=True)
|
except (KeyError, RuntimeError) as exc:
|
raise ValueError("Model weights do not match the supported PointWiseNet architecture.") from exc
|
model.eval()
|
return model, codes
|
|
|
def main() -> int:
|
parser = argparse.ArgumentParser(description="Apply a local point-cloud semantic model to a new RGB point cloud.")
|
parser.add_argument("--model", type=Path, required=True)
|
parser.add_argument("--input", type=Path, required=True)
|
parser.add_argument("--output", type=Path, required=True)
|
parser.add_argument("--device", choices={"cpu", "cuda"}, default="cpu")
|
parser.add_argument("--batch-size", type=int, default=4096)
|
args = parser.parse_args()
|
if args.device == "cuda" and not torch.cuda.is_available():
|
raise SystemExit("CUDA was requested but is unavailable.")
|
if args.batch_size < 1 or args.batch_size > 262_144:
|
raise SystemExit("Batch size must be between 1 and 262144.")
|
if not args.model.is_file() or not args.input.is_file():
|
raise SystemExit("Model or input point cloud is unavailable.")
|
if args.output.exists() and any(args.output.iterdir()):
|
raise SystemExit("Output directory must be new or empty.")
|
|
started = time.perf_counter()
|
device = torch.device(args.device)
|
model, class_codes = load_model(args.model, device)
|
xyz, rgb, source_las = load_cloud(args.input)
|
center = xyz.mean(axis=0)
|
scale = float(max(np.abs(xyz - center).max(), 1e-6))
|
features = np.column_stack(((xyz - center) / scale, rgb)).astype(np.float32)
|
pieces: list[np.ndarray] = []
|
with torch.no_grad():
|
for start in range(0, len(features), args.batch_size):
|
values = torch.from_numpy(features[start:start + args.batch_size]).to(device)
|
pieces.append(model(values).argmax(dim=1).cpu().numpy())
|
predictions = np.asarray([class_codes[index] for index in np.concatenate(pieces)], dtype=np.uint8)
|
counts = {int(code): int((predictions == code).sum()) for code in class_codes}
|
|
args.output.mkdir(parents=True, exist_ok=True)
|
preview_selection = preview_indices(predictions, PREVIEW_POINT_LIMIT)
|
preview = o3d.geometry.PointCloud()
|
preview.points = o3d.utility.Vector3dVector(xyz[preview_selection])
|
preview.colors = o3d.utility.Vector3dVector(np.asarray([CLASS_SCHEMA[int(code)]["color"] for code in predictions[preview_selection]], dtype=np.float64) / 255.0)
|
preview_path = args.output / "predicted-semantic-preview.ply"
|
if not o3d.io.write_point_cloud(str(preview_path), preview, write_ascii=False):
|
raise RuntimeError("Could not write predicted PLY preview.")
|
|
classified_las = args.output / "predicted-semantic-classified.las"
|
if source_las is not None:
|
source_las.classification = predictions
|
source_las.write(classified_las)
|
else:
|
header = laspy.LasHeader(point_format=3, version="1.2")
|
header.offsets = xyz.min(axis=0)
|
header.scales = np.array([0.001, 0.001, 0.001])
|
generated = laspy.LasData(header)
|
generated.x, generated.y, generated.z = xyz[:, 0], xyz[:, 1], xyz[:, 2]
|
generated.red = np.rint(rgb[:, 0] * 65535).astype(np.uint16)
|
generated.green = np.rint(rgb[:, 1] * 65535).astype(np.uint16)
|
generated.blue = np.rint(rgb[:, 2] * 65535).astype(np.uint16)
|
generated.classification = predictions
|
generated.write(classified_las)
|
|
csv_path = args.output / "class-counts.csv"
|
with csv_path.open("w", newline="", encoding="utf-8") as stream:
|
writer = csv.DictWriter(stream, fieldnames=["class_code", "class_key", "class_label", "point_count"])
|
writer.writeheader()
|
for code in class_codes:
|
writer.writerow({"class_code": code, "class_key": CLASS_SCHEMA[code]["key"], "class_label": CLASS_SCHEMA[code]["label"], "point_count": counts[code]})
|
summary = {"class_codes": class_codes, "class_counts": {str(code): counts[code] for code in class_codes}, "input_points": int(len(xyz)), "preview_points": int(len(preview_selection)), "preview_sampling": "deterministic class-aware cap; smaller predicted classes retained before the remaining budget is sampled", "input_has_rgb": True}
|
summary_path = args.output / "prediction-summary.json"
|
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
metadata = {"capability": "05-3d-pointcloud", "classification": "B", "created_at": datetime.now(UTC).isoformat(), "model": {"path": str(args.model), "sha256": sha256(args.model), "architecture": "PointWiseNet shared MLP"}, "input": {"path": str(args.input), "sha256": sha256(args.input), "bytes": args.input.stat().st_size, "points": int(len(xyz)), "has_rgb": True}, "classes": {str(code): CLASS_SCHEMA[code] for code in class_codes}, "prediction": summary, "processing": {"device": args.device, "batch_size": args.batch_size, "normalization": {"method": "source-local per input", "xyz_center": center.tolist(), "xyz_scale": scale}}, "versions": {"python": sys.version.split()[0], "torch": torch.__version__, "open3d": o3d.__version__, "laspy": laspy.__version__}, "artifacts": {"preview": preview_path.name, "classified_las": classified_las.name, "class_counts": csv_path.name, "summary": summary_path.name}, "elapsed_seconds": round(time.perf_counter() - started, 3), "limitations": ["Predictions are model candidates, not asset inventory or inspection conclusions.", "This model requires observed RGB; it cannot infer labels for XYZ-only point clouds.", "Model metrics apply only to the labelled source spatial blocks. The current pole/tower class has high false-positive risk and requires review.", "New inputs are normalized with their own XYZ centre and scale to match the training feature definition; this preserves their coordinates but does not prove cross-site generalization."]}
|
(args.output / "run_metadata.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
print(json.dumps(metadata, ensure_ascii=False))
|
return 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
