From 45bbcc48ae44ba671e25293698a2e9124fd1cf6e Mon Sep 17 00:00:00 2001 From: Juho Date: Fri, 23 Oct 2026 16:00:01 +0300 Subject: [PATCH 1/2] feat: add parallel module --- src/geogenalg/core/parallel.py | 309 +++++++++++++ .../unit/core/test_parallel_benchmark.py | 215 +++++++++ tests/unit/core/test_parallel.py | 435 ++++++++++++++++++ whitelist.txt | 7 + 4 files changed, 966 insertions(+) create mode 100644 src/geogenalg/core/parallel.py create mode 100644 tests/bench/unit/core/test_parallel_benchmark.py create mode 100644 tests/unit/core/test_parallel.py diff --git a/src/geogenalg/core/parallel.py b/src/geogenalg/core/parallel.py new file mode 100644 index 00000000..21ec57e6 --- /dev/null +++ b/src/geogenalg/core/parallel.py @@ -0,0 +1,309 @@ +# Copyright (c) 2025 National Land Survey of Finland (Maanmittauslaitos) +# +# This file is part of geogen-algorithms. +# +# SPDX-License-Identifier: MIT + +from __future__ import annotations + +import atexit +import os +import threading +from concurrent.futures import ThreadPoolExecutor +from typing import TYPE_CHECKING, Any, Literal +from warnings import warn + +import numpy as np +from geopandas import GeoSeries +from geopandas.array import GeometryArray +from pandas import Index, Series +from shapely import ( + BufferCapStyle, + BufferJoinStyle, + buffer, + difference, + intersection, + symmetric_difference, +) + +if TYPE_CHECKING: + from collections.abc import Callable + + from pyproj import CRS + from shapely.geometry.base import BaseGeometry + +GEOGENALG_WORKER_COUNT_ENV_VAR = "GEOGENALG_NUM_WORKERS" + + +class _ThreadManager: + """Private singleton class for managing a thread pool executor.""" + + def __init__(self) -> None: + self._num_workers: int | None = None + self._executor: ThreadPoolExecutor | None = None + self._mutex = threading.Lock() + + atexit.register(self.shutdown) + + def get_worker_count(self) -> int: + # Prefer explicitly set number + if self._num_workers is not None: + return self._num_workers + + # Env variable as a first fallback + if (env_variable := os.getenv(GEOGENALG_WORKER_COUNT_ENV_VAR)) is not None: + try: + as_int = int(env_variable) + if as_int > 0: + return as_int + warn( + f"{GEOGENALG_WORKER_COUNT_ENV_VAR} must be > 0. Using default.", + UserWarning, + stacklevel=3, + ) + except ValueError: + warn( + f"{GEOGENALG_WORKER_COUNT_ENV_VAR} could not be parsed as an " + "integer. Using default.", + UserWarning, + stacklevel=3, + ) + # No explicit setting or env variable -> try to use number of cores + + # Only in >= Python 3.13 + if hasattr(os, "process_cpu_count") and (count := os.process_cpu_count()): + return count + + # For Python 3.12 (UNIX only) + if hasattr(os, "sched_getaffinity"): + try: + return len(os.sched_getaffinity(0)) + except OSError: + pass + + return os.cpu_count() or 1 + + @property + def executor(self) -> ThreadPoolExecutor: + with self._mutex: + if self._executor is None: + self._executor = ThreadPoolExecutor(max_workers=self.get_worker_count()) + + return self._executor + + @property + def num_workers(self) -> int: + return self.get_worker_count() + + @num_workers.setter + def num_workers(self, value: int | None) -> None: + if value is not None and value < 1: + msg = "Value must be >= 1." + raise ValueError(msg) + + with self._mutex: + self._num_workers = value + + if self._executor is not None: + self._executor.shutdown(wait=False) + self._executor = ThreadPoolExecutor(max_workers=self.get_worker_count()) + + def shutdown(self) -> None: + with self._mutex: + if self._executor is not None: + self._executor.shutdown(wait=True) + self._executor = None + + +_THREAD_MANAGER = _ThreadManager() + + +def num_workers() -> int: + """Return number of workers.""" + return _THREAD_MANAGER.num_workers + + +def set_num_workers(value: int | None) -> None: + """Set number of workers and reconfigure thread pool.""" + _THREAD_MANAGER.num_workers = value + + +def get_threadpool_executor() -> ThreadPoolExecutor: + """Get thread pool executor.""" + return _THREAD_MANAGER.executor + + +def shutdown_threadpool_executor() -> None: + """Manually shutdown thread pool executor.""" + _THREAD_MANAGER.shutdown() + + +def _get_chunk_slices(value_count: int, chunk_count: int) -> list[slice]: + chunk_size, remainder = divmod(value_count, chunk_count) + stops = [i * chunk_size + min(i, remainder) for i in range(chunk_count + 1)] + return [slice(stops[i], stops[i + 1]) for i in range(chunk_count)] + + +def _process_arg( + value: object, + _slice: slice, + total_geom_count: int, +) -> slice | object: + if isinstance(value, np.ndarray) and len(value) == total_geom_count: + return value[_slice] + + if isinstance(value, Series) and len(value) == total_geom_count: + return value.iloc[_slice] + + return value + + +def _process_geometry_array( + geoms: np.ndarray, + geom_function: Callable[..., Any], + *args: object, + executor: ThreadPoolExecutor | None = None, + **kwargs: object, +) -> np.ndarray: + geom_count = len(geoms) + if geom_count == 0: + return np.array([], dtype="object") + + workers = min(num_workers(), geom_count) + if workers <= 1: + return np.asarray(geom_function(geoms, *args, **kwargs)) + + def worker(_slice: slice) -> np.ndarray: + worker_args = [_process_arg(arg, _slice, geom_count) for arg in args] + worker_kwargs = { + key: _process_arg(value, _slice, geom_count) + for key, value in kwargs.items() + } + return np.asarray( + geom_function(geoms[_slice], *worker_args, **worker_kwargs), + ) + + pool = executor or get_threadpool_executor() + return np.concatenate( + list(pool.map(worker, _get_chunk_slices(geom_count, workers))), + ) + + +class ParallelGeometryProcess: + """Convenience class to allow chaining geometry operations. + + All operations are are executed in parallel (depending on the number of + workers configured.) + """ + + def __init__( + self, + data: np.ndarray, + index: Index, + crs: str | int | tuple[str, str] | list[str] | CRS | None, + executor: ThreadPoolExecutor | None = None, + ) -> None: + """Initialize object.""" + self._data = data + self._index = index + self._crs = crs + self._executor = executor + + def run( + self, + func: Callable[..., Any], + *args: object, + **kwargs: object, + ) -> ParallelGeometryProcess: + """Run geometry processing function in parallel.""" + processed_args = [ + a.to_numpy() if isinstance(a, (Series, GeoSeries)) else a for a in args + ] + + processed_kwargs = { + k: (v.to_numpy() if isinstance(v, (Series, GeoSeries)) else v) + for k, v in kwargs.items() + } + + result = _process_geometry_array( + self._data, + func, + *processed_args, + executor=self._executor, + **processed_kwargs, + ) + + return ParallelGeometryProcess( + result, + self._index, + self._crs, + self._executor, + ) + + def buffer( # noqa: PLR0913 + self, + distance: float | np.ndarray | Series, + resolution: int = 16, + cap_style: BufferCapStyle | Literal["round", "square", "flat"] = "round", + join_style: BufferJoinStyle | Literal["round", "mitre", "bevel"] = "round", + mitre_limit: float = 5.0, + *, + single_sided: bool = False, + ) -> ParallelGeometryProcess: + """Run shapely buffer in parallel.""" + return self.run( + buffer, + distance, + quad_segs=resolution, + cap_style=cap_style, + join_style=join_style, + mitre_limit=mitre_limit, + single_sided=single_sided, + ) + + def intersection( + self, + other: BaseGeometry | np.ndarray | GeoSeries, + grid_size: float | None = None, + ) -> ParallelGeometryProcess: + """Run shapely intersection in parallel.""" + return self.run(intersection, other, grid_size=grid_size) + + def difference( + self, + other: BaseGeometry | np.ndarray | GeoSeries, + grid_size: float | None = None, + ) -> ParallelGeometryProcess: + """Run shapely difference in parallel.""" + return self.run(difference, other, grid_size=grid_size) + + def symmetric_difference( + self, + other: BaseGeometry | np.ndarray | GeoSeries, + grid_size: float | None = None, + ) -> ParallelGeometryProcess: + """Run shapely symmetric difference in parallel.""" + return self.run(symmetric_difference, other, grid_size=grid_size) + + def resolve_parallel(self) -> Series | GeoSeries: + """Resolve process back into pandas object.""" + if self._data.dtype == object: + return GeoSeries( + GeometryArray(self._data, crs=self._crs), + index=self._index, + ) + + return Series(self._data, index=self._index) + + +def in_parallel( + geoms: GeoSeries, + executor: ThreadPoolExecutor | None = None, +) -> ParallelGeometryProcess: + """Create geometry process object from GeoSeries.""" + return ParallelGeometryProcess( + data=geoms.to_numpy(), + index=geoms.index, + crs=geoms.crs, + executor=executor, + ) diff --git a/tests/bench/unit/core/test_parallel_benchmark.py b/tests/bench/unit/core/test_parallel_benchmark.py new file mode 100644 index 00000000..4c1ce5a3 --- /dev/null +++ b/tests/bench/unit/core/test_parallel_benchmark.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from typing import TYPE_CHECKING + +import pytest +from geopandas import GeoDataFrame, GeoSeries +from geopandas.testing import assert_geoseries_equal +from pandas import Series +from pandas.testing import assert_series_equal +from shapely import area, box, buffer, intersection, intersects, simplify + +from geogenalg.core.parallel import in_parallel, num_workers + +if TYPE_CHECKING: + from collections.abc import Callable + + import numpy as np + from pytest_benchmark.fixture import BenchmarkFixture + + +@pytest.mark.benchmark +@pytest.mark.parametrize("feature_count", [10, 100, 1_000, 10_000]) +@pytest.mark.parametrize( + "mode", ["no_pool", "threadmanager_pool", "non_cached_threadpool"] +) +def test_benchmark_buffer( + benchmark: BenchmarkFixture, + make_point_gdf: Callable[..., GeoDataFrame], + feature_count: int, + mode: str, +) -> None: + geoms = make_point_gdf(feature_count).geometry + + if mode == "no_pool": + + def run_process() -> GeoSeries: + return GeoSeries(buffer(geoms.to_numpy(), 5.0, quad_segs=16), crs=geoms.crs) + + elif mode == "threadmanager_pool": + + def run_process() -> Series | GeoSeries: + return in_parallel(geoms).buffer(5.0).resolve_parallel() + + else: # non_cached_threadpool + + def run_process() -> Series | GeoSeries: + with ThreadPoolExecutor(max_workers=num_workers()) as fresh_pool: + return ( + in_parallel(geoms, executor=fresh_pool) + .buffer(5.0) + .resolve_parallel() + ) + + result = benchmark(run_process) + assert_geoseries_equal(result, geoms.buffer(5.0)) + + +@pytest.mark.benchmark +@pytest.mark.parametrize("feature_count", [10, 100, 1_000, 10_000]) +@pytest.mark.parametrize( + "mode", ["no_pool", "threadmanager_pool", "non_cached_threadpool"] +) +def test_benchmark_intersects_single_geom( + benchmark: BenchmarkFixture, + make_overlapping_polygons: Callable[..., GeoDataFrame], + feature_count: int, + mode: str, +) -> None: + geoms = make_overlapping_polygons(feature_count).geometry + target_box = box(0.0, 0.0, 50.0, 50.0) + + if mode == "no_pool": + + def run_process() -> Series: + return Series(intersects(geoms.to_numpy(), target_box)) + + elif mode == "threadmanager_pool": + + def run_process() -> Series | GeoSeries: + return in_parallel(geoms).run(intersects, target_box).resolve_parallel() + + else: # non_cached_threadpool + + def run_process() -> Series | GeoSeries: + with ThreadPoolExecutor(max_workers=num_workers()) as fresh_pool: + return ( + in_parallel(geoms, executor=fresh_pool) + .run(intersects, target_box) + .resolve_parallel() + ) + + result = benchmark(run_process) + assert_series_equal(result, geoms.intersects(target_box)) + + +@pytest.mark.benchmark +@pytest.mark.parametrize("feature_count", [10, 100, 1_000, 10_000]) +@pytest.mark.parametrize( + "mode", ["no_pool", "threadmanager_pool", "non_cached_threadpool"] +) +def test_benchmark_intersection_pairwise( + benchmark: BenchmarkFixture, + make_overlapping_polygons: Callable[..., GeoDataFrame], + feature_count: int, + mode: str, +) -> None: + geoms_1 = make_overlapping_polygons(feature_count, seed=0).geometry + geoms_2 = geoms_1.translate(xoff=0.5, yoff=0.5) + + if mode == "no_pool": + + def run_process() -> GeoSeries: + return GeoSeries(intersection(geoms_1, geoms_2), crs=geoms_1.crs) + + elif mode == "threadmanager_pool": + + def run_process() -> Series | GeoSeries: + return in_parallel(geoms_1).intersection(geoms_2).resolve_parallel() + + else: # non_cached_threadpool + + def run_process() -> Series | GeoSeries: + with ThreadPoolExecutor(max_workers=num_workers()) as fresh_pool: + return ( + in_parallel(geoms_1, executor=fresh_pool) + .intersection(geoms_2) + .resolve_parallel() + ) + + result = benchmark(run_process) + assert_geoseries_equal(result, geoms_1.intersection(geoms_2)) + + +@pytest.mark.benchmark +@pytest.mark.parametrize("feature_count", [10, 100, 1_000, 10_000]) +@pytest.mark.parametrize( + "mode", ["no_pool", "threadmanager_pool", "non_cached_threadpool"] +) +def test_benchmark_area( + benchmark: BenchmarkFixture, + make_overlapping_polygons: Callable[..., GeoDataFrame], + feature_count: int, + mode: str, +) -> None: + geoms = make_overlapping_polygons(feature_count).geometry + + if mode == "no_pool": + + def run_process() -> np.ndarray: + return area(geoms.to_numpy()) + + elif mode == "threadmanager_pool": + + def run_process() -> Series | GeoSeries: + return in_parallel(geoms).run(area).resolve_parallel() + + else: # non_cached_threadpool + + def run_process() -> Series | GeoSeries: + with ThreadPoolExecutor(max_workers=num_workers()) as fresh_pool: + return ( + in_parallel(geoms, executor=fresh_pool).run(area).resolve_parallel() + ) + + result = benchmark(run_process) + assert len(result) == feature_count + + +@pytest.mark.benchmark +@pytest.mark.parametrize("feature_count", [10, 100, 1000, 10_000]) +@pytest.mark.parametrize( + "mode", + ["no_pool", "threadmanager_pool", "non_cached_threadpool"], +) +def test_process_with_multiple_steps( + benchmark: BenchmarkFixture, + make_point_gdf: Callable[..., GeoDataFrame], + feature_count: int, + mode: str, +): + geoms = make_point_gdf(feature_count).geometry + + if mode == "no_pool": + + def run_process() -> np.ndarray: + step1 = buffer(geoms.to_numpy(), 5.0, quad_segs=16) + step2 = simplify(step1, tolerance=0.5) + return buffer(step2, 5.0) + + elif mode == "threadmanager_pool": + + def run_process() -> Series | GeoSeries: + return ( + in_parallel(geoms) + .buffer(5.0) + .run(simplify, tolerance=0.5) + .buffer(5.0) + .resolve_parallel() + ) + + else: # non_cached_threadpool + + def run_process() -> Series | GeoSeries: + with ThreadPoolExecutor(max_workers=num_workers()) as fresh_pool: + return ( + in_parallel(geoms, executor=fresh_pool) + .buffer(5.0) + .run(simplify, tolerance=0.5) + .buffer(5.0) + .resolve_parallel() + ) + + result = benchmark(run_process) + assert len(result) == feature_count diff --git a/tests/unit/core/test_parallel.py b/tests/unit/core/test_parallel.py new file mode 100644 index 00000000..14fdb167 --- /dev/null +++ b/tests/unit/core/test_parallel.py @@ -0,0 +1,435 @@ +# Copyright (c) 2025 National Land Survey of Finland (Maanmittauslaitos) +# +# This file is part of geogen-algorithms. +# +# SPDX-License-Identifier: MIT + +from __future__ import annotations + +import re +import threading +from concurrent.futures import ThreadPoolExecutor +from contextlib import nullcontext +from typing import TYPE_CHECKING, cast + +import numpy as np +import pytest +from geopandas import GeoSeries +from geopandas.testing import assert_geoseries_equal +from pandas import Index, Series +from shapely import ( + box, + buffer, + equals, + length, +) +from shapely.geometry import Point + +from geogenalg.core.parallel import ( + _THREAD_MANAGER, + GEOGENALG_WORKER_COUNT_ENV_VAR, + ParallelGeometryProcess, + _get_chunk_slices, + _process_arg, + _process_geometry_array, + get_threadpool_executor, + in_parallel, + num_workers, + set_num_workers, + shutdown_threadpool_executor, +) + +if TYPE_CHECKING: + from collections.abc import Generator + + +@pytest.fixture(autouse=True) +def reset_thread_manager() -> Generator[None, None, None]: + shutdown_threadpool_executor() + set_num_workers(None) + yield + shutdown_threadpool_executor() + set_num_workers(None) + + +@pytest.fixture +def sample_points() -> GeoSeries: + geoms = [Point(i, i) for i in range(12)] + index = Index([f"idx_{i}" for i in range(12)], name="index") + return GeoSeries(geoms, index=index, crs="EPSG:3857") + + +@pytest.mark.parametrize( + ("value_count", "chunk_count", "expected"), + [ + (10, 2, [slice(0, 5), slice(5, 10)]), + (10, 3, [slice(0, 4), slice(4, 7), slice(7, 10)]), + (5, 5, [slice(0, 1), slice(1, 2), slice(2, 3), slice(3, 4), slice(4, 5)]), + (1, 1, [slice(0, 1)]), + (0, 1, [slice(0, 0)]), + ], +) +def test_get_chunk_slices( + value_count: int, + chunk_count: int, + expected: list[slice], +): + slices = _get_chunk_slices(value_count, chunk_count) + assert slices == expected + total_elements = sum(s.stop - s.start for s in slices) + assert total_elements == value_count + + +@pytest.mark.parametrize( + ("arg", "total_geom_count", "expected"), + [ + pytest.param( + np.array([10, 20, 30, 40]), + 4, + [np.array([10]), np.array([20]), np.array([30]), np.array([40])], + id="array_gets_sliced", + ), + pytest.param( + np.array([10, 20, 30, 40]), + 10, + [np.array([10, 20, 30, 40])] * 4, + id="array_gets_broadcasted", + ), + pytest.param( + Series([10, 20, 30, 40], index=["a", "b", "c", "d"]), + 4, + [ + Series([10], index=["a"]), + Series([20], index=["b"]), + Series([30], index=["c"]), + Series([40], index=["d"]), + ], + id="series_gets_sliced", + ), + pytest.param( + Series([10, 20]), + 4, + [Series([10, 20])] * 4, + id="series_gets_broadcasted", + ), + pytest.param( + 5.0, + 4, + [5.0, 5.0, 5.0, 5.0], + id="scalar_float", + ), + pytest.param( + "round", + 4, + ["round", "round", "round", "round"], + id="scalar_string", + ), + pytest.param( + None, + 4, + [None, None, None, None], + id="scalar_none", + ), + ], +) +def test_process_arg( + arg: object, + total_geom_count: int, + expected: list[object], +): + for i in range(len(expected)): + _slice = slice(i, i + 1) + result = _process_arg(arg, _slice, total_geom_count) + + if isinstance(result, np.ndarray): + np.testing.assert_array_equal(result, expected[i]) + elif isinstance(result, Series): + assert result.equals(cast("Series", expected[i])) + else: + assert result == expected[i] + + +def test_worker_count_setter(): + set_num_workers(4) + assert num_workers() == 4 + assert get_threadpool_executor()._max_workers == 4 + + set_num_workers(2) + assert num_workers() == 2 + assert get_threadpool_executor()._max_workers == 2 + + +def test_worker_count_setter_raises(): + with pytest.raises(ValueError, match=re.escape("Value must be >= 1.")): + set_num_workers(0) + + with pytest.raises(ValueError, match=re.escape("Value must be >= 1.")): + set_num_workers(-5) + + +def test_worker_count_env_variable(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv(GEOGENALG_WORKER_COUNT_ENV_VAR, "3") + set_num_workers(None) + assert num_workers() == 3 + + +@pytest.mark.parametrize( + ("env_var", "warn_message"), + [ + ("0", "must be > 0. Using default."), + ("text", "could not be parsed as an integer. Using default."), + ], + ids=[ + "invalid_number", + "not_a_number", + ], +) +def test_worker_count_env_variable_warns( + monkeypatch: pytest.MonkeyPatch, + env_var: str, + warn_message: str, +): + monkeypatch.setenv(GEOGENALG_WORKER_COUNT_ENV_VAR, env_var) + set_num_workers(None) + with pytest.warns(UserWarning, match=re.escape(warn_message)): + assert num_workers() >= 1 + + +def test_executor_lifecycle_and_shutdown(): + executor = get_threadpool_executor() + assert isinstance(executor, ThreadPoolExecutor) + assert not executor._shutdown + + shutdown_threadpool_executor() + assert _THREAD_MANAGER._executor is None + + new_executor = get_threadpool_executor() + assert new_executor is not executor + assert not new_executor._shutdown + + +def test_worker_setter_recreates_pool(): + executor1 = get_threadpool_executor() + set_num_workers(3) + executor2 = get_threadpool_executor() + + assert executor1 is not executor2 + assert executor1._shutdown + assert executor2._max_workers == 3 + + +@pytest.mark.parametrize( + ( + "geoms", + "workers", + "use_custom_executor", + "expect_parallel", + ), + [ + pytest.param( + np.array([], dtype=object), + 4, + False, + False, + id="empty_input", + ), + pytest.param( + np.array([Point(0, 0)], dtype=object), + 4, + False, + False, + id="single_input_runs_on_main", + ), + pytest.param( + np.array([Point(0, 0), Point(1, 1)], dtype=object), + 1, + False, + False, + id="set_single_worker_runs_on_main", + ), + pytest.param( + np.array([Point(i, i) for i in range(4)], dtype=object), + 2, + False, + True, + id="four_geoms_get_split_to_two_workers", + ), + pytest.param( + np.array([Point(i, i) for i in range(7)], dtype=object), + 3, + False, + True, + id="uneven_split_to_workers", + ), + pytest.param( + np.array([Point(0, 0), Point(1, 1), Point(2, 2)], dtype=object), + 8, + False, + True, + id="more_workers_than_geoms", + ), + pytest.param( + np.array([Point(i, i) for i in range(10)], dtype=object), + 4, + True, + True, + id="custom_executor", + ), + ], +) +def test_process_geometry_array( + geoms: np.ndarray, + workers: int, + use_custom_executor: bool, + expect_parallel: bool, +): + set_num_workers(workers) + + main_thread_id = threading.get_ident() + thread_ids: set[int] = set() + lock = threading.Lock() + + def tracking_buffer(geoms: np.ndarray, distance: float) -> np.ndarray: + with lock: + thread_ids.add(threading.get_ident()) + return buffer(geoms, distance) + + pool_context = ( + ThreadPoolExecutor(max_workers=workers) + if use_custom_executor + else nullcontext() + ) + + with pool_context as custom_executor: + result = _process_geometry_array( + geoms, + tracking_buffer, + 1, + executor=custom_executor, + ) + + assert isinstance(result, np.ndarray) + assert len(result) == len(geoms) + assert result.dtype == object + + expected = buffer(geoms, 1) + assert equals(result, expected).all() + + if len(geoms) == 0: + assert len(thread_ids) == 0 + elif expect_parallel: + assert len(thread_ids) >= 1 + assert main_thread_id not in thread_ids + else: + assert thread_ids == {main_thread_id} + + +def test_in_parallel(sample_points: GeoSeries): + process = in_parallel(sample_points) + + assert isinstance(process, ParallelGeometryProcess) + assert len(process._data) == len(sample_points) + + +def test_resolve_parallel(sample_points: GeoSeries): + process = in_parallel(sample_points).resolve_parallel() + + assert isinstance(process, GeoSeries) + assert_geoseries_equal(process, sample_points) + + process = in_parallel(sample_points).run(length).resolve_parallel() + + assert isinstance(process, Series) + assert (process == 0.0).all() + + +def test_buffer_with_scalar_distance(sample_points: GeoSeries): + set_num_workers(4) + result = in_parallel(sample_points).buffer(1.5).resolve_parallel() + + assert isinstance(result, GeoSeries) + assert result.crs == sample_points.crs + assert result.index.equals(sample_points.index) + + expected = buffer(sample_points, 1.5, quad_segs=16) + assert equals(result, expected).all() + + +def test_buffer_with_non_scalar_distance(sample_points: GeoSeries): + set_num_workers(4) + distances = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]) + + result = in_parallel(sample_points).buffer(distances).resolve_parallel() + expected = buffer(sample_points, distances, quad_segs=16) + + assert equals(result, expected).all() + + +def test_intersection_against_single_geom(sample_points: GeoSeries): + poly = box(-0.1, -0.1, 0.1, 0.1) + + result = in_parallel(sample_points).intersection(poly).resolve_parallel() + + assert result.iloc[0] == Point(0, 0) + + for i in range(1, 12): + assert result.iloc[i].is_empty + + +def test_difference_against_single_geom(sample_points: GeoSeries): + poly = box(-0.1, -0.1, 0.1, 0.1) + + result = in_parallel(sample_points).difference(poly).resolve_parallel() + + assert result.iloc[0].is_empty + + for i in range(1, 12): + assert result.iloc[i] == Point(i, i) + + +def test_intersection_pairwise(sample_points: GeoSeries): + boxes = np.array( + [box(i - 0.1, i - 0.1, i + 0.1, i + 0.1) for i in range(len(sample_points))], + dtype=object, + ) + + result = in_parallel(sample_points).intersection(boxes).resolve_parallel() + + for i in range(len(sample_points)): + assert result.iloc[i] == Point(i, i) + + +def test_difference_pairwise(sample_points: GeoSeries): + boxes = np.array( + [box(i - 0.1, i - 0.1, i + 0.1, i + 0.1) for i in range(len(sample_points))], + dtype=object, + ) + + result = in_parallel(sample_points).difference(boxes).resolve_parallel() + + for i in range(len(sample_points)): + assert result.iloc[i].is_empty + + +def test_in_parallel_chaining(sample_points: GeoSeries): + result = in_parallel(sample_points).buffer(1.0).buffer(5.0).resolve_parallel() + + assert isinstance(result, GeoSeries) + + expected = sample_points.buffer(1.0).buffer(5.0) + + assert_geoseries_equal(result, expected) + + +def test_in_parallel_chaining_non_cached_pool(sample_points: GeoSeries): + with ThreadPoolExecutor(max_workers=2) as pool: + result = ( + in_parallel(sample_points, executor=pool) + .buffer(1.0) + .buffer(5.0) + .resolve_parallel() + ) + + assert isinstance(result, GeoSeries) + expected = sample_points.buffer(1.0).buffer(5.0) + assert_geoseries_equal(result, expected) diff --git a/whitelist.txt b/whitelist.txt index 76cf10f4..74f52f3a 100644 --- a/whitelist.txt +++ b/whitelist.txt @@ -13,6 +13,7 @@ argspec asarray ascontiguousarray atan2 +atexit atol basealgorithm basemetadata @@ -66,6 +67,7 @@ geopackage geopackages geopandas geoseries +getaffinity getfullargspec getmro getsource @@ -94,6 +96,7 @@ multigeometries multilinestring multilinestrings multipolygon +mutex ndarray networkx nodata @@ -105,6 +108,7 @@ pos postprocessing pydantic pygeoops +pyproj ramer rdp rect @@ -113,10 +117,12 @@ rglob rtol Rtree ruas +sched scipy searchsorted segmentize segmentized +segs shapelysmooth shlex simplifier @@ -135,6 +141,7 @@ subsample substring substrings testdata +threadpool tol tolist topo From ed80cff6dd03fa02255e0d91cc117f5dfd6ec06e Mon Sep 17 00:00:00 2001 From: Juho Date: Mon, 26 Oct 2026 10:15:11 +0200 Subject: [PATCH 2/2] docs: add note about instantiation --- src/geogenalg/core/parallel.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/geogenalg/core/parallel.py b/src/geogenalg/core/parallel.py index 21ec57e6..ada46cf7 100644 --- a/src/geogenalg/core/parallel.py +++ b/src/geogenalg/core/parallel.py @@ -194,6 +194,8 @@ class ParallelGeometryProcess: All operations are are executed in parallel (depending on the number of workers configured.) + + Objects are intended to be instantiated with the in_parallel() function. """ def __init__(