will/osm-import-tools
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Speed up address and road diffs from hours to seconds, gzip GeoJSON downloads, and render map points on canvas with viewport-only OSM/county layers

Browse Source
This commit is contained in:
zyphlar
2026-10-21 19:46:02 -07:00
parent 28550984be
commit 4759424e85
4 changed files with 296 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files
diff-highways.py
+60 -167
View File
@@ -30,7 +30,6 @@ import warnings
import multiprocessing as mp
from functools import partial
import numpy as np
from concurrent.futures import ProcessPoolExecutor, as_completed
import gc
import importlib
@@ -72,7 +71,6 @@ class RoadComparator:
self.tolerance_deg = tolerance_feet / 364000.0
self.min_gap_length_deg = min_gap_length_feet / 364000.0
print(f"Using {self.n_jobs} parallel processes with chunk size {self.chunk_size}")
if self.exclude_unnamed:
print("Excluding unnamed features from coverage calculation")
@@ -132,61 +130,28 @@ class RoadComparator:
except Exception as e:
raise Exception(f"Error loading {filepath}: {str(e)}")
def create_buffered_union_optimized(self, gdf: gpd.GeoDataFrame) -> Any:
"""Create a buffered union using chunked processing for memory efficiency."""
print("Creating optimized buffered union...")
# Process in chunks to manage memory - extract geometries as lists
chunks = [gdf.iloc[i:i+self.chunk_size].geometry.tolist() for i in range(0, len(gdf), self.chunk_size)]
chunk_unions = []
# Use partial function for multiprocessing
buffer_func = partial(self._buffer_chunk, tolerance=self.tolerance_deg)
with ProcessPoolExecutor(max_workers=self.n_jobs) as executor:
# Submit all chunk processing jobs
future_to_chunk = {executor.submit(buffer_func, chunk): i
for i, chunk in enumerate(chunks)}
# Collect results as they complete
for future in as_completed(future_to_chunk):
chunk_idx = future_to_chunk[future]
try:
chunk_union = future.result(timeout=300) # 5 minute timeout
if chunk_union and not chunk_union.is_empty:
chunk_unions.append(chunk_union)
print(f"Processed chunk {chunk_idx + 1}/{len(chunks)}")
except Exception as e:
print(f"Error processing chunk {chunk_idx}: {str(e)}")
# Continue processing other chunks instead of failing completely
# Union all chunk results
print("Combining chunk unions...")
if chunk_unions:
final_union = unary_union(chunk_unions)
# Force garbage collection
del chunk_unions, chunks
gc.collect()
return final_union
else:
raise Exception("No valid geometries to create union")
def create_buffer_index(self, gdf: gpd.GeoDataFrame) -> Tuple[np.ndarray, STRtree]:
"""Buffer every geometry and index the buffers.
Each road is later compared only against the union of the buffers that
touch it (see _local_cover), which gives the same difference as using
one county-wide union - buffers that don't touch a line can't change
its difference - without every comparison paying for the size of the
whole county."""
print("Buffering and indexing geometries...")
buffers = np.asarray(gdf.geometry.buffer(self.tolerance_deg).values, dtype=object)
return buffers, STRtree(buffers)
@staticmethod
def _buffer_chunk(geometries: List, tolerance: float) -> Any:
"""Buffer geometries in a chunk and return their union."""
try:
# Buffer all geometries in the chunk
buffered = [geom.buffer(tolerance) for geom in geometries]
# Create union of buffered geometries
if len(buffered) == 1:
return buffered[0]
else:
return unary_union(buffered)
except Exception as e:
print(f"Error in chunk processing: {str(e)}")
def _local_cover(geom, buffers: np.ndarray, tree: STRtree):
"""Union of the target buffers that touch geom, or None if none do."""
idx = tree.query(geom, predicate='intersects')
if len(idx) == 0:
return None
if len(idx) == 1:
return buffers[idx[0]]
return unary_union(list(buffers[idx]))
def create_spatial_index(self, gdf: gpd.GeoDataFrame) -> STRtree:
"""Create spatial index for fast intersection queries."""
print("Creating spatial index...")
@@ -194,57 +159,30 @@ class RoadComparator:
geometries = gdf.geometry.tolist()
return STRtree(geometries)
def find_removed_segments_optimized(self, source_gdf: gpd.GeoDataFrame,
target_union: Any) -> List[Dict[str, Any]]:
def find_removed_segments_optimized(self, source_gdf: gpd.GeoDataFrame,
target: Tuple[np.ndarray, STRtree]) -> List[Dict[str, Any]]:
"""
Find segments in source_gdf that are not covered by target_union (removed roads).
Optimized with parallel processing.
Find segments in source_gdf that are not covered by the target buffers (removed roads).
"""
print("Finding removed segments...")
# Split into chunks for parallel processing - convert to serializable format
chunks = []
for i in range(0, len(source_gdf), self.chunk_size):
chunk_gdf = source_gdf.iloc[i:i+self.chunk_size]
chunk_data = []
for idx, row in chunk_gdf.iterrows():
chunk_data.append({
'geometry': row.geometry,
'properties': dict(row.drop('geometry'))
})
chunks.append(chunk_data)
all_removed = []
# Use partial function for multiprocessing
process_func = partial(self._process_removed_chunk,
target_union=target_union,
min_length_deg=self.min_gap_length_deg)
with ProcessPoolExecutor(max_workers=self.n_jobs) as executor:
# Submit all chunk processing jobs
future_to_chunk = {executor.submit(process_func, chunk): i
for i, chunk in enumerate(chunks)}
# Collect results as they complete
for future in as_completed(future_to_chunk):
chunk_idx = future_to_chunk[future]
try:
chunk_removed = future.result(timeout=300) # 5 minute timeout
all_removed.extend(chunk_removed)
print(f"Processed removed chunk {chunk_idx + 1}/{len(chunks)}")
except Exception as e:
print(f"Error processing removed chunk {chunk_idx}: {str(e)}")
# Continue processing other chunks instead of failing completely
# Clean up memory
del chunks
gc.collect()
return all_removed
rows = self._rows_with_cover(source_gdf, target, "removed")
return self._process_removed_chunk(rows, self.min_gap_length_deg)
def _rows_with_cover(self, gdf: gpd.GeoDataFrame, target: Tuple[np.ndarray, STRtree],
label: str) -> List[Dict[str, Any]]:
buffers, tree = target
geoms = gdf.geometry.tolist()
props = gdf.drop(columns='geometry').to_dict('records')
rows = []
for i, (geom, p) in enumerate(zip(geoms, props)):
rows.append({'geometry': geom, 'properties': p,
'cover': self._local_cover(geom, buffers, tree)})
if (i + 1) % 10000 == 0:
print(f" {label}: compared {i + 1}/{len(geoms)}")
return rows
@staticmethod
def _process_removed_chunk(chunk_data: List[Dict], target_union: Any,
min_length_deg: float) -> List[Dict[str, Any]]:
def _process_removed_chunk(chunk_data: List[Dict], min_length_deg: float) -> List[Dict[str, Any]]:
"""Process a chunk of geometries to find removed segments."""
removed_segments = []
@@ -261,7 +199,8 @@ class RoadComparator:
for line in lines:
try:
# Find parts of the line that don't intersect with target_union
uncovered = line.difference(target_union)
cover = row_data['cover']
uncovered = line.difference(cover) if cover is not None else line
if uncovered.is_empty:
continue
@@ -292,57 +231,17 @@ class RoadComparator:
return removed_segments
def find_added_roads_optimized(self, source_gdf: gpd.GeoDataFrame,
target_union: Any) -> List[Dict[str, Any]]:
def find_added_roads_optimized(self, source_gdf: gpd.GeoDataFrame,
target: Tuple[np.ndarray, STRtree]) -> List[Dict[str, Any]]:
"""
Find entire roads in source_gdf that don't significantly overlap with target_union.
Optimized with parallel processing.
Find entire roads in source_gdf that don't significantly overlap with the target buffers.
"""
print("Finding added roads...")
# Split into chunks for parallel processing - convert to serializable format
chunks = []
for i in range(0, len(source_gdf), self.chunk_size):
chunk_gdf = source_gdf.iloc[i:i+self.chunk_size]
chunk_data = []
for idx, row in chunk_gdf.iterrows():
chunk_data.append({
'geometry': row.geometry,
'properties': dict(row.drop('geometry'))
})
chunks.append(chunk_data)
all_added = []
# Use partial function for multiprocessing
process_func = partial(self._process_added_chunk,
target_union=target_union,
min_length_deg=self.min_gap_length_deg)
with ProcessPoolExecutor(max_workers=self.n_jobs) as executor:
# Submit all chunk processing jobs
future_to_chunk = {executor.submit(process_func, chunk): i
for i, chunk in enumerate(chunks)}
# Collect results as they complete
for future in as_completed(future_to_chunk):
chunk_idx = future_to_chunk[future]
try:
chunk_added = future.result(timeout=300) # 5 minute timeout
all_added.extend(chunk_added)
print(f"Processed added chunk {chunk_idx + 1}/{len(chunks)}")
except Exception as e:
print(f"Error processing added chunk {chunk_idx}: {str(e)}")
# Continue processing other chunks instead of failing completely
# Clean up memory
del chunks
gc.collect()
return all_added
rows = self._rows_with_cover(source_gdf, target, "added")
return self._process_added_chunk(rows, self.min_gap_length_deg)
@staticmethod
def _process_added_chunk(chunk_data: List[Dict], target_union: Any,
min_length_deg: float) -> List[Dict[str, Any]]:
def _process_added_chunk(chunk_data: List[Dict], min_length_deg: float) -> List[Dict[str, Any]]:
"""Process a chunk of geometries to find added roads."""
added_roads = []
@@ -352,7 +251,8 @@ class RoadComparator:
try:
# Check what portion of the road is not covered
uncovered = geom.difference(target_union)
cover = row_data['cover']
uncovered = geom.difference(cover) if cover is not None else geom
if not uncovered.is_empty:
# Calculate what percentage of the original road is uncovered
@@ -465,7 +365,6 @@ class RoadComparator:
print(f"Comparing {file1_path} and {file2_path}")
print(f"Tolerance: {self.tolerance_feet} feet")
print(f"Minimum significant length: {self.min_gap_length_feet} feet")
print(f"Parallel processing: {self.n_jobs} workers")
print("-" * 50)
# Load both files
@@ -493,20 +392,14 @@ class RoadComparator:
print(f"Warning: CRS mismatch. Converting {file2_path} to match {file1_path}")
gdf2 = gdf2.to_crs(gdf1.crs)
print("Creating optimized spatial unions...")
# Create buffered unions using optimized method
union1 = self.create_buffered_union_optimized(gdf1)
union2 = self.create_buffered_union_optimized(gdf2)
print("Finding removed and added roads with parallel processing...")
# Find roads using optimized parallel methods
removed_roads = self.find_removed_segments_optimized(gdf1, union2)
added_roads = self.find_added_roads_optimized(gdf2, union1)
index1 = self.create_buffer_index(gdf1)
index2 = self.create_buffer_index(gdf2)
removed_roads = self.find_removed_segments_optimized(gdf1, index2)
added_roads = self.find_added_roads_optimized(gdf2, index1)
# Clean up memory
del gdf1, gdf2, union1, union2
del gdf1, gdf2, index1, index2
gc.collect()
return removed_roads, added_roads
@@ -635,9 +528,9 @@ Examples:
help='Minimum length in feet for gaps/extras to be considered significant (default: 100)')
parser.add_argument('--output', '-o', help='Output GeoJSON file for results (optional)')
parser.add_argument('--jobs', '-j', type=int, default=None,
help='Number of parallel processes (default: CPU count - 1)')
help='Ignored (kept for compatibility)')
parser.add_argument('--chunk-size', '-c', type=int, default=1000,
help='Number of geometries to process per chunk (default: 1000)')
help='Ignored (kept for compatibility)')
parser.add_argument('--exclude-unnamed', '-e', action='store_true',
help='Exclude features without name tags from coverage calculation (helps detect roads covered by unnamed geometry)')