mytec
defa3ad440
Major refactoring of RFCP backend: - Modular propagation models (8 models) - SharedMemoryManager for terrain data - ProcessPoolExecutor parallel processing - WebSocket progress streaming - Building filtering pipeline (351k → 15k) - 82 unit tests Performance: Standard preset 38s → 5s (7.6x speedup) Known issue: Detailed preset timeout (fix in 3.1.0)
117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
"""
|
|
Vectorized line-segment and line-polygon intersection checks.
|
|
|
|
All operations use NumPy for batch processing.
|
|
"""
|
|
|
|
import numpy as np
|
|
from typing import Tuple
|
|
|
|
|
|
def line_segments_intersect_batch(
|
|
p1: np.ndarray, p2: np.ndarray,
|
|
segments_start: np.ndarray, segments_end: np.ndarray,
|
|
) -> Tuple[np.ndarray, np.ndarray]:
|
|
"""Check if line p1->p2 intersects with N segments.
|
|
|
|
Args:
|
|
p1, p2: shape (2,)
|
|
segments_start, segments_end: shape (N, 2)
|
|
|
|
Returns:
|
|
intersects: bool array (N,)
|
|
t_values: parameter along p1->p2 (N,)
|
|
"""
|
|
d = p2 - p1
|
|
seg_d = segments_end - segments_start
|
|
|
|
cross = d[0] * seg_d[:, 1] - d[1] * seg_d[:, 0]
|
|
parallel_mask = np.abs(cross) < 1e-10
|
|
cross_safe = np.where(parallel_mask, 1.0, cross)
|
|
|
|
dp = p1 - segments_start
|
|
t = (dp[:, 0] * seg_d[:, 1] - dp[:, 1] * seg_d[:, 0]) / cross_safe
|
|
u = (dp[:, 0] * d[1] - dp[:, 1] * d[0]) / cross_safe
|
|
|
|
intersects = ~parallel_mask & (t >= 0) & (t <= 1) & (u >= 0) & (u <= 1)
|
|
return intersects, t
|
|
|
|
|
|
def line_intersects_polygons_batch(
|
|
p1: np.ndarray, p2: np.ndarray,
|
|
polygons_x: np.ndarray, polygons_y: np.ndarray,
|
|
polygon_lengths: np.ndarray,
|
|
max_polygons: int = 30,
|
|
) -> Tuple[np.ndarray, np.ndarray]:
|
|
"""Check if line p1->p2 intersects multiple polygons.
|
|
|
|
Uses bounding-box pre-filter to limit work when polygon count is large.
|
|
|
|
Args:
|
|
p1, p2: shape (2,)
|
|
polygons_x, polygons_y: flattened vertex arrays
|
|
polygon_lengths: vertices per polygon (num_polygons,)
|
|
max_polygons: only check nearest N polygons
|
|
|
|
Returns:
|
|
intersects: bool (num_polygons,)
|
|
min_distances: distance to first hit (num_polygons,)
|
|
"""
|
|
num_polygons = len(polygon_lengths)
|
|
|
|
if num_polygons == 0:
|
|
return np.array([], dtype=bool), np.array([])
|
|
|
|
intersects = np.zeros(num_polygons, dtype=bool)
|
|
min_t = np.full(num_polygons, np.inf)
|
|
|
|
# Pre-filter: bounding box check
|
|
if num_polygons > max_polygons:
|
|
buf = 50.0
|
|
line_min_x = min(p1[0], p2[0]) - buf
|
|
line_max_x = max(p1[0], p2[0]) + buf
|
|
line_min_y = min(p1[1], p2[1]) - buf
|
|
line_max_y = max(p1[1], p2[1]) + buf
|
|
|
|
nearby_mask = np.zeros(num_polygons, dtype=bool)
|
|
vi = 0
|
|
for i, length in enumerate(polygon_lengths):
|
|
if length >= 3:
|
|
cx = polygons_x[vi]
|
|
cy = polygons_y[vi]
|
|
if line_min_x <= cx <= line_max_x and line_min_y <= cy <= line_max_y:
|
|
nearby_mask[i] = True
|
|
vi += length
|
|
|
|
nearby_indices = np.where(nearby_mask)[0]
|
|
if len(nearby_indices) > max_polygons:
|
|
nearby_mask = np.zeros(num_polygons, dtype=bool)
|
|
nearby_mask[nearby_indices[:max_polygons]] = True
|
|
else:
|
|
nearby_mask = np.ones(num_polygons, dtype=bool)
|
|
|
|
idx = 0
|
|
for i, length in enumerate(polygon_lengths):
|
|
if length < 3 or not nearby_mask[i]:
|
|
idx += length
|
|
continue
|
|
|
|
px = polygons_x[idx:idx + length]
|
|
py = polygons_y[idx:idx + length]
|
|
|
|
starts = np.stack([px, py], axis=1)
|
|
ends = np.stack([np.roll(px, -1), np.roll(py, -1)], axis=1)
|
|
|
|
edge_intersects, t_vals = line_segments_intersect_batch(p1, p2, starts, ends)
|
|
|
|
if np.any(edge_intersects):
|
|
intersects[i] = True
|
|
min_t[i] = np.min(t_vals[edge_intersects])
|
|
|
|
idx += length
|
|
|
|
line_length = np.linalg.norm(p2 - p1)
|
|
min_distances = min_t * line_length
|
|
|
|
return intersects, min_distances
|