"""Parametric scene generator: objects, terrain surfaces, intersection clipping, export.""" from __future__ import annotations import math import random from typing import Any # --------------------------------------------------------------------------- # Catalog / default params # --------------------------------------------------------------------------- LAYER_CATALOG: list[dict[str, Any]] = [ { "kind": "object", "type": "pipe", "label": "Труба", "params": [ {"key": "length", "label": "Длина", "type": "number", "default": 2.6, "min": 0.2, "max": 20, "step": 0.1}, {"key": "radius", "label": "Радиус", "type": "number", "default": 0.22, "min": 0.02, "max": 5, "step": 0.01}, {"key": "axis", "label": "Ось", "type": "select", "default": "y", "options": ["x", "y", "z"]}, {"key": "count", "label": "Точек", "type": "number", "default": 2500, "min": 100, "max": 100000, "step": 100}, {"key": "noise", "label": "Шум", "type": "number", "default": 0.01, "min": 0, "max": 0.5, "step": 0.005}, {"key": "seed", "label": "Seed", "type": "number", "default": 1, "min": 0, "max": 999999, "step": 1}, ], }, { "kind": "object", "type": "sphere", "label": "Сфера", "params": [ {"key": "radius", "label": "Радиус", "type": "number", "default": 0.5, "min": 0.05, "max": 10, "step": 0.05}, {"key": "count", "label": "Точек", "type": "number", "default": 2000, "min": 100, "max": 100000, "step": 100}, {"key": "noise", "label": "Шум", "type": "number", "default": 0.02, "min": 0, "max": 0.5, "step": 0.005}, {"key": "seed", "label": "Seed", "type": "number", "default": 1, "min": 0, "max": 999999, "step": 1}, ], }, { "kind": "object", "type": "box", "label": "Параллелепипед", "params": [ {"key": "sizeX", "label": "Размер X", "type": "number", "default": 1.0, "min": 0.1, "max": 20, "step": 0.1}, {"key": "sizeY", "label": "Размер Y", "type": "number", "default": 0.6, "min": 0.1, "max": 20, "step": 0.1}, {"key": "sizeZ", "label": "Размер Z", "type": "number", "default": 0.4, "min": 0.1, "max": 20, "step": 0.1}, {"key": "count", "label": "Точек", "type": "number", "default": 2000, "min": 100, "max": 100000, "step": 100}, {"key": "noise", "label": "Шум", "type": "number", "default": 0.01, "min": 0, "max": 0.5, "step": 0.005}, {"key": "seed", "label": "Seed", "type": "number", "default": 1, "min": 0, "max": 999999, "step": 1}, ], }, { "kind": "object", "type": "torus", "label": "Тор", "params": [ {"key": "majorR", "label": "Большой R", "type": "number", "default": 1.0, "min": 0.1, "max": 10, "step": 0.05}, {"key": "minorR", "label": "Малый R", "type": "number", "default": 0.35, "min": 0.02, "max": 5, "step": 0.01}, {"key": "count", "label": "Точек", "type": "number", "default": 2500, "min": 100, "max": 100000, "step": 100}, {"key": "noise", "label": "Шум", "type": "number", "default": 0.01, "min": 0, "max": 0.5, "step": 0.005}, {"key": "seed", "label": "Seed", "type": "number", "default": 1, "min": 0, "max": 999999, "step": 1}, ], }, { "kind": "surface", "type": "ocean_floor", "label": "Дно океана", "params": [ {"key": "sizeX", "label": "Размер X", "type": "number", "default": 4.0, "min": 0.5, "max": 50, "step": 0.1}, {"key": "sizeY", "label": "Размер Y", "type": "number", "default": 3.0, "min": 0.5, "max": 50, "step": 0.1}, {"key": "amplitude", "label": "Амплитуда", "type": "number", "default": 0.12, "min": 0, "max": 2, "step": 0.01}, {"key": "frequency", "label": "Частота", "type": "number", "default": 2.2, "min": 0.1, "max": 20, "step": 0.1}, {"key": "channel", "label": "Канал", "type": "number", "default": 0.08, "min": 0, "max": 1, "step": 0.01}, {"key": "baseZ", "label": "База Z", "type": "number", "default": -0.45, "min": -20, "max": 20, "step": 0.05}, {"key": "count", "label": "Точек", "type": "number", "default": 4000, "min": 100, "max": 100000, "step": 100}, {"key": "noise", "label": "Шум", "type": "number", "default": 0.02, "min": 0, "max": 0.5, "step": 0.005}, {"key": "seed", "label": "Seed", "type": "number", "default": 1, "min": 0, "max": 999999, "step": 1}, ], }, { "kind": "surface", "type": "wave", "label": "Волна", "params": [ {"key": "sizeX", "label": "Размер X", "type": "number", "default": 2.4, "min": 0.5, "max": 50, "step": 0.1}, {"key": "sizeY", "label": "Размер Y", "type": "number", "default": 2.4, "min": 0.5, "max": 50, "step": 0.1}, {"key": "amplitude", "label": "Амплитуда", "type": "number", "default": 0.35, "min": 0, "max": 5, "step": 0.05}, {"key": "frequency", "label": "Частота", "type": "number", "default": 2.5, "min": 0.1, "max": 20, "step": 0.1}, {"key": "count", "label": "Точек", "type": "number", "default": 3000, "min": 100, "max": 100000, "step": 100}, {"key": "noise", "label": "Шум", "type": "number", "default": 0.015, "min": 0, "max": 0.5, "step": 0.005}, {"key": "seed", "label": "Seed", "type": "number", "default": 1, "min": 0, "max": 999999, "step": 1}, ], }, { "kind": "surface", "type": "flat", "label": "Плоскость", "params": [ {"key": "sizeX", "label": "Размер X", "type": "number", "default": 4.0, "min": 0.5, "max": 50, "step": 0.1}, {"key": "sizeY", "label": "Размер Y", "type": "number", "default": 4.0, "min": 0.5, "max": 50, "step": 0.1}, {"key": "z", "label": "Высота Z", "type": "number", "default": -0.5, "min": -20, "max": 20, "step": 0.05}, {"key": "count", "label": "Точек", "type": "number", "default": 2500, "min": 100, "max": 100000, "step": 100}, {"key": "noise", "label": "Шум", "type": "number", "default": 0.01, "min": 0, "max": 0.5, "step": 0.005}, {"key": "seed", "label": "Seed", "type": "number", "default": 1, "min": 0, "max": 999999, "step": 1}, ], }, ] _CATALOG_BY_KEY = {(item["kind"], item["type"]): item for item in LAYER_CATALOG} LAYER_COLORS = { ("object", "pipe"): "#f59e0b", ("object", "sphere"): "#38bdf8", ("object", "box"): "#a78bfa", ("object", "torus"): "#34d399", ("surface", "ocean_floor"): "#64748b", ("surface", "wave"): "#94a3b8", ("surface", "flat"): "#78716c", } def catalog_payload() -> dict[str, Any]: return {"layers": LAYER_CATALOG, "colors": {f"{k[0]}:{k[1]}": v for k, v in LAYER_COLORS.items()}} def default_params(kind: str, type_name: str) -> dict[str, Any]: entry = _CATALOG_BY_KEY.get((kind, type_name)) if entry is None: raise ValueError(f"Unknown layer type: {kind}/{type_name}") return {p["key"]: p["default"] for p in entry["params"]} def merge_params(kind: str, type_name: str, params: dict[str, Any] | None) -> dict[str, Any]: merged = default_params(kind, type_name) if params: for key, value in params.items(): if key in merged: merged[key] = value # Coerce numeric fields entry = _CATALOG_BY_KEY[(kind, type_name)] for p in entry["params"]: key = p["key"] if p["type"] == "number" and key in merged: try: merged[key] = float(merged[key]) if key in ("count", "seed"): merged[key] = int(merged[key]) except (TypeError, ValueError): merged[key] = p["default"] if p["type"] == "select" and key in merged: options = p.get("options") or [] if merged[key] not in options: merged[key] = p["default"] return merged # --------------------------------------------------------------------------- # Generation # --------------------------------------------------------------------------- def _jitter(rng: random.Random, noise: float) -> float: if noise <= 0: return 0.0 return rng.uniform(-noise, noise) def generate_pipe(params: dict[str, Any]) -> list[list[float]]: rng = random.Random(int(params["seed"])) count = max(1, int(params["count"])) length = float(params["length"]) radius = float(params["radius"]) noise = float(params["noise"]) axis = params.get("axis", "y") points: list[list[float]] = [] half = length * 0.5 for _ in range(count): angle = rng.random() * 2.0 * math.pi t = rng.uniform(-half, half) radial = radius + _jitter(rng, noise) cx = radial * math.cos(angle) cy = radial * math.sin(angle) if axis == "x": points.append([t, cx, cy]) elif axis == "z": points.append([cx, cy, t]) else: points.append([cx, t, cy]) return points def generate_sphere(params: dict[str, Any]) -> list[list[float]]: rng = random.Random(int(params["seed"])) count = max(1, int(params["count"])) radius = float(params["radius"]) noise = float(params["noise"]) points: list[list[float]] = [] for _ in range(count): u = rng.uniform(-1.0, 1.0) theta = rng.random() * 2.0 * math.pi r = radius + _jitter(rng, noise) s = math.sqrt(max(0.0, 1.0 - u * u)) points.append([r * s * math.cos(theta), r * s * math.sin(theta), r * u]) return points def generate_box(params: dict[str, Any]) -> list[list[float]]: """Sample points on the box surface.""" rng = random.Random(int(params["seed"])) count = max(1, int(params["count"])) sx = float(params["sizeX"]) * 0.5 sy = float(params["sizeY"]) * 0.5 sz = float(params["sizeZ"]) * 0.5 noise = float(params["noise"]) faces = [ ("x", sx, sy, sz), ("x", -sx, sy, sz), ("y", sy, sx, sz), ("y", -sy, sx, sz), ("z", sz, sx, sy), ("z", -sz, sx, sy), ] areas = [abs(a[2]) * abs(a[3]) * 4.0 for a in faces] total = sum(areas) or 1.0 points: list[list[float]] = [] for _ in range(count): pick = rng.random() * total acc = 0.0 face = faces[0] for f, area in zip(faces, areas): acc += area if pick <= acc: face = f break axis, fixed, u_max, v_max = face u = rng.uniform(-u_max, u_max) v = rng.uniform(-v_max, v_max) jx, jy, jz = _jitter(rng, noise), _jitter(rng, noise), _jitter(rng, noise) if axis == "x": points.append([fixed + jx, u + jy, v + jz]) elif axis == "y": points.append([u + jx, fixed + jy, v + jz]) else: points.append([u + jx, v + jy, fixed + jz]) return points def generate_torus(params: dict[str, Any]) -> list[list[float]]: rng = random.Random(int(params["seed"])) count = max(1, int(params["count"])) major_r = float(params["majorR"]) minor_r = float(params["minorR"]) noise = float(params["noise"]) points: list[list[float]] = [] for _ in range(count): u = rng.random() * 2.0 * math.pi v = rng.random() * 2.0 * math.pi radial = minor_r + _jitter(rng, noise) x = (major_r + radial * math.cos(v)) * math.cos(u) y = (major_r + radial * math.cos(v)) * math.sin(u) z = radial * math.sin(v) points.append([x, y, z]) return points def height_ocean_floor(x: float, y: float, params: dict[str, Any]) -> float: amplitude = float(params["amplitude"]) frequency = float(params["frequency"]) channel = float(params["channel"]) base_z = float(params["baseZ"]) waviness = amplitude * math.cos(frequency * y) channel_term = channel * x * x return base_z + channel_term + waviness def height_wave(x: float, y: float, params: dict[str, Any]) -> float: amplitude = float(params["amplitude"]) frequency = float(params["frequency"]) return amplitude * math.sin(frequency * x) * math.cos(frequency * y) def height_flat(_x: float, _y: float, params: dict[str, Any]) -> float: return float(params["z"]) def surface_height_fn(type_name: str): if type_name == "ocean_floor": return height_ocean_floor if type_name == "wave": return height_wave if type_name == "flat": return height_flat raise ValueError(f"Unknown surface type: {type_name}") def generate_surface(type_name: str, params: dict[str, Any]) -> list[list[float]]: rng = random.Random(int(params["seed"])) count = max(1, int(params["count"])) size_x = float(params.get("sizeX", 2.0)) size_y = float(params.get("sizeY", 2.0)) noise = float(params["noise"]) height_fn = surface_height_fn(type_name) half_x = size_x * 0.5 half_y = size_y * 0.5 points: list[list[float]] = [] for _ in range(count): x = rng.uniform(-half_x, half_x) y = rng.uniform(-half_y, half_y) z = height_fn(x, y, params) + _jitter(rng, noise) points.append([x, y, z]) return points _GENERATORS = { ("object", "pipe"): generate_pipe, ("object", "sphere"): generate_sphere, ("object", "box"): generate_box, ("object", "torus"): generate_torus, } def generate_layer(kind: str, type_name: str, params: dict[str, Any] | None = None) -> dict[str, Any]: if (kind, type_name) not in _CATALOG_BY_KEY: raise ValueError(f"Unknown layer type: {kind}/{type_name}") merged = merge_params(kind, type_name, params) if kind == "surface": points = generate_surface(type_name, merged) else: points = _GENERATORS[(kind, type_name)](merged) entry = _CATALOG_BY_KEY[(kind, type_name)] return { "kind": kind, "type": type_name, "label": entry["label"], "params": merged, "pointCount": len(points), "points": points, "color": LAYER_COLORS.get((kind, type_name), "#7dd3fc"), } # --------------------------------------------------------------------------- # Transforms & intersections # --------------------------------------------------------------------------- def normalize_transform(transform: dict[str, Any] | None) -> dict[str, float]: t = transform or {} return { "x": float(t.get("x", 0.0) or 0.0), "y": float(t.get("y", 0.0) or 0.0), "z": float(t.get("z", 0.0) or 0.0), "rx": float(t.get("rx", 0.0) or 0.0), "ry": float(t.get("ry", 0.0) or 0.0), "rz": float(t.get("rz", 0.0) or 0.0), } def _rotate_xyz(x: float, y: float, z: float, rx: float, ry: float, rz: float) -> tuple[float, float, float]: """Euler XYZ (same as Three.js Object3D.rotation default order).""" cx, sx = math.cos(rx), math.sin(rx) cy, sy = math.cos(ry), math.sin(ry) cz, sz = math.cos(rz), math.sin(rz) y, z = y * cx - z * sx, y * sx + z * cx x, z = x * cy + z * sy, -x * sy + z * cy x, y = x * cz - y * sz, x * sz + y * cz return x, y, z def _rotate_xyz_inverse(x: float, y: float, z: float, rx: float, ry: float, rz: float) -> tuple[float, float, float]: cx, sx = math.cos(rx), math.sin(rx) cy, sy = math.cos(ry), math.sin(ry) cz, sz = math.cos(rz), math.sin(rz) x, y = x * cz + y * sz, -x * sz + y * cz x, z = x * cy - z * sy, x * sy + z * cy y, z = y * cx + z * sx, -y * sx + z * cx return x, y, z def apply_transform(points: list[list[float]], transform: dict[str, Any] | None) -> list[list[float]]: t = normalize_transform(transform) out: list[list[float]] = [] for p in points: x, y, z = _rotate_xyz(p[0], p[1], p[2], t["rx"], t["ry"], t["rz"]) out.append([x + t["x"], y + t["y"], z + t["z"]]) return out def _world_to_local(point: list[float], transform: dict[str, Any] | None) -> tuple[float, float, float]: t = normalize_transform(transform) x = point[0] - t["x"] y = point[1] - t["y"] z = point[2] - t["z"] return _rotate_xyz_inverse(x, y, z, t["rx"], t["ry"], t["rz"]) def object_sdf(type_name: str, params: dict[str, Any], local: tuple[float, float, float]) -> float: """Signed distance: negative = inside.""" if type_name == "imported": # No analytic SDF for imported clouds — skip solid clipping. return 1.0 x, y, z = local if type_name == "sphere": return math.sqrt(x * x + y * y + z * z) - float(params["radius"]) if type_name == "pipe": radius = float(params["radius"]) half = float(params["length"]) * 0.5 axis = params.get("axis", "y") if axis == "x": radial = math.sqrt(y * y + z * z) - radius axial = abs(x) - half elif axis == "z": radial = math.sqrt(x * x + y * y) - radius axial = abs(z) - half else: radial = math.sqrt(x * x + z * z) - radius axial = abs(y) - half # Approximate solid cylinder: inside if radial < 0 and axial < 0 outside = max(radial, axial) if radial < 0 and axial < 0: return max(radial, axial) if axial > 0 and radial < 0: return axial if radial > 0 and axial < 0: return radial return math.sqrt(max(radial, 0) ** 2 + max(axial, 0) ** 2) if outside > 0 else outside if type_name == "box": hx = float(params["sizeX"]) * 0.5 hy = float(params["sizeY"]) * 0.5 hz = float(params["sizeZ"]) * 0.5 qx = abs(x) - hx qy = abs(y) - hy qz = abs(z) - hz outside = math.sqrt(max(qx, 0) ** 2 + max(qy, 0) ** 2 + max(qz, 0) ** 2) inside = min(max(qx, qy, qz), 0.0) return outside + inside if type_name == "torus": major_r = float(params["majorR"]) minor_r = float(params["minorR"]) q = math.sqrt(x * x + y * y) - major_r return math.sqrt(q * q + z * z) - minor_r return 1.0 def point_below_surface( world_pt: list[float], surf_type: str, surf_params: dict[str, Any], surf_transform: dict[str, Any] | None, eps: float, ) -> bool: """True if world point is below the heightfield in the surface local frame.""" lx, ly, lz = _world_to_local(world_pt, surf_transform) size_x = float(surf_params.get("sizeX", 1e9)) size_y = float(surf_params.get("sizeY", 1e9)) if abs(lx) > size_x * 0.5 + eps or abs(ly) > size_y * 0.5 + eps: return False h = surface_height_fn(surf_type)(lx, ly, surf_params) return lz < h + eps def resolve_intersections( layers: list[dict[str, Any]], *, eps: float = 0.01, clip_surface_inside_objects: bool = True, clip_objects_vs_objects: bool = True, ) -> list[dict[str, Any]]: """Return layers with points updated (local coords preserved via inverse transform).""" prepared: list[dict[str, Any]] = [] for layer in layers: kind = layer["kind"] type_name = layer["type"] transform = normalize_transform(layer.get("transform")) local_points = layer.get("points") if type_name == "imported": params = dict(layer.get("params") or {}) if not local_points: raise ValueError("Imported layer has no points.") label = layer.get("label") or "OBJ" color = layer.get("color") or "#f472b6" else: params = merge_params(kind, type_name, layer.get("params")) if not local_points: generated = generate_layer(kind, type_name, params) local_points = generated["points"] label = layer.get("label") or _CATALOG_BY_KEY[(kind, type_name)]["label"] color = layer.get("color") or LAYER_COLORS.get((kind, type_name), "#7dd3fc") world = apply_transform(local_points, transform) prepared.append({ "id": layer.get("id"), "kind": kind, "type": type_name, "params": params, "transform": transform, "local_points": local_points, "world_points": world, "color": color, "label": label, }) surfaces = [p for p in prepared if p["kind"] == "surface"] objects = [p for p in prepared if p["kind"] == "object"] result: list[dict[str, Any]] = [] for layer in prepared: keep_local: list[list[float]] = [] keep_world: list[list[float]] = [] for local_pt, world_pt in zip(layer["local_points"], layer["world_points"]): drop = False if layer["kind"] == "object": for surf in surfaces: if point_below_surface( world_pt, surf["type"], surf["params"], surf["transform"], eps ): drop = True break if not drop and clip_objects_vs_objects: for other in objects: if other is layer: continue local_in_other = _world_to_local(world_pt, other["transform"]) if object_sdf(other["type"], other["params"], local_in_other) < -eps: drop = True break elif layer["kind"] == "surface" and clip_surface_inside_objects: for obj in objects: local_in_obj = _world_to_local(world_pt, obj["transform"]) if object_sdf(obj["type"], obj["params"], local_in_obj) < -eps: drop = True break if not drop: keep_local.append([local_pt[0], local_pt[1], local_pt[2]]) keep_world.append(world_pt) result.append({ "id": layer["id"], "kind": layer["kind"], "type": layer["type"], "label": layer["label"], "params": layer["params"], "transform": layer["transform"], "color": layer["color"], "pointCount": len(keep_local), "points": keep_local, "removedCount": len(layer["local_points"]) - len(keep_local), }) return result def merge_layers_world(layers: list[dict[str, Any]]) -> list[list[float]]: merged: list[list[float]] = [] for layer in layers: kind = layer["kind"] type_name = layer["type"] transform = normalize_transform(layer.get("transform")) points = layer.get("points") if not points: if type_name == "imported": continue params = merge_params(kind, type_name, layer.get("params")) points = generate_layer(kind, type_name, params)["points"] merged.extend(apply_transform(points, transform)) return merged def layer_semantic_class(layer: dict[str, Any]) -> float: """Binary PointNet label: 1 = pipe, 0 = everything else.""" type_name = str(layer.get("type") or "").lower() if type_name == "pipe": return 1.0 # Optional name hint for renamed imported clouds name = str(layer.get("name") or layer.get("label") or "").lower() if "pipe" in name or "труб" in name: return 1.0 return 0.0 def points_to_pointnet_rows(points: list[list[float]], class_label: float) -> list[list[float]]: """XYZRGB+class rows; RGB forced to 0; float values (stored as float64 in .npy).""" c = float(class_label) rows: list[list[float]] = [] for p in points: rows.append([float(p[0]), float(p[1]), float(p[2]), 0.0, 0.0, 0.0, c]) return rows def layers_to_pointnet_rows(layers: list[dict[str, Any]]) -> list[list[float]]: rows: list[list[float]] = [] for layer in layers: kind = layer.get("kind") or "object" type_name = layer.get("type") or "imported" transform = normalize_transform(layer.get("transform")) points = layer.get("points") if not points: if type_name == "imported": continue params = merge_params(kind, type_name, layer.get("params")) points = generate_layer(kind, type_name, params)["points"] world = apply_transform(points, transform) rows.extend(points_to_pointnet_rows(world, layer_semantic_class(layer))) return rows def export_npy_float64(rows: list[list[float]]) -> bytes: """Write NumPy .npy v1.0 binary array shape (N, C) dtype float64 little-endian.""" import struct n = len(rows) cols = len(rows[0]) if n else 7 if n and any(len(r) != cols for r in rows): raise ValueError("All rows must have the same length for .npy export.") header = "{'descr': ' str: return "\n".join(f"{p[0]:.8f} {p[1]:.8f} {p[2]:.8f}" for p in points) + ("\n" if points else "") def export_ply(points: list[list[float]]) -> str: header = ( "ply\n" "format ascii 1.0\n" f"element vertex {len(points)}\n" "property float x\n" "property float y\n" "property float z\n" "end_header\n" ) body = "\n".join(f"{p[0]:.8f} {p[1]:.8f} {p[2]:.8f}" for p in points) return header + body + ("\n" if points else "") def export_obj(points: list[list[float]], object_name: str = "cloud") -> str: safe_name = "".join(ch if ch.isalnum() or ch in "_-" else "_" for ch in (object_name or "cloud")) or "cloud" lines = [f"# DotsToSurface point cloud ({len(points)} vertices)", f"o {safe_name}"] for p in points: lines.append(f"v {p[0]:.8f} {p[1]:.8f} {p[2]:.8f}") return "\n".join(lines) + "\n" def parse_obj_points(text: str) -> list[list[float]]: """Extract vertex positions from Wavefront OBJ (ignores faces/materials).""" points: list[list[float]] = [] for raw in text.splitlines(): line = raw.strip() if not line or line.startswith("#"): continue if line.lower().startswith("v "): parts = line.split() if len(parts) < 4: continue try: points.append([float(parts[1]), float(parts[2]), float(parts[3])]) except ValueError: continue return points