add some tests, draft sound parallel cfg, minor bugfixes

cases/sound_waves/configuration/config_parallel.py

@@ -0,0 +1,207 @@
+from copy import deepcopy
+from pathlib import Path
+
+import numpy as np
+
+from cases.sound_waves.microphone_points import Microphone
+from gefest.tools.utils import poly_from_comsol_txt
+from gefest.core.configs.optimization_params import OptimizationParams
+from gefest.core.configs.tuner_params import TunerParams
+from gefest.core.geometry.datastructs.structure import Structure
+from gefest.core.geometry.domain import Domain
+from gefest.core.geometry.utils import get_random_structure
+from gefest.core.opt.objective.objective import Objective
+from gefest.tools.estimators.simulators.sound_wave.sound_interface import (
+    SoundSimulator,
+    generate_map,
+)
+
+# # # Metrics # # #
+DAMPING = 1 - 0.001
+CA2 = 0.5
+INITIAL_P = 200
+MAX_PRESSURE = INITIAL_P / 2
+MIN_PRESSURE = -INITIAL_P / 2
+
+import itertools
+from numpy.core.umath import pi
+class SoundFieldFitness(Objective):
+    """Evaluates sound pressure level difference with reference."""
+
+    def __init__(self, domain, estimator, best_spl, duration):
+        super().__init__(domain, estimator)
+        self.best_spl = best_spl
+        self.omega = 3 / (2 * pi)
+        self.iteration = 0
+        self.map_size = (round(1.2 * domain.max_y), round(1.2 * domain.max_x))
+        self.size_y, self.size_x = self.map_size
+        self.duration = duration
+        # obstacle_map handling
+
+        # Source position is the center of the map
+        self.s_y = self.size_y // 2
+        self.s_x = self.size_x // 2
+
+    def update_velocity(self, velocities, pressure, obstacle_map):
+        """Update the velocity field based on Komatsuzaki's transition rules."""
+        V = velocities
+        P = pressure
+        for i, j in itertools.product(range(self.size_y), range(self.size_x)):
+            if obstacle_map[i, j] == 1:
+                V[i, j, 0:4] = 0.0
+                continue
+
+            V[i, j, 0] = V[i, j, 0] + P[i, j] - P[i - 1, j] if i > 0 else P[i, j]
+            V[i, j, 1] = V[i, j, 1] + P[i, j] - P[i, j + 1] if j < self.size_x - 1 else P[i, j]
+            V[i, j, 2] = V[i, j, 2] + P[i, j] - P[i + 1, j] if i < self.size_y - 1 else P[i, j]
+            V[i, j, 3] = V[i, j, 3] + P[i, j] - P[i, j - 1] if j > 0 else P[i, j]
+
+        return V
+
+    def step(self, velocities, pressure, obstacle_map):
+        """Perform a simulation step, upadting the wind an pressure fields."""
+        pressure[self.s_y, self.s_x] = INITIAL_P * np.sin(self.omega * self.iteration)
+        velocities = self.update_velocity(velocities, pressure, obstacle_map)
+        pressure = self.update_perssure(pressure, velocities)
+        self.iteration += 1
+        return velocities, pressure
+
+    def update_perssure(self, pressure, velocities):
+        """Update the pressure field based on Komatsuzaki's transition rules."""
+        pressure -= CA2 * DAMPING * np.sum(velocities, axis=2)
+        return pressure
+
+    def spl(self, pressure_hist, integration_interval=60):
+        """Computes the sound pressure level map.
+
+        https://en.wikipedia.org/wiki/Sound_pressure#Sound_pressure_level
+
+        Args:
+            integration_interval (int): interval over which the rms pressure
+                                        is computed, starting from the last
+                                        simulation iteration backwards.
+
+        Returns:
+            spl (np.array): map of sound pressure level (dB).
+        """
+        p0 = 20 * 10e-6  # Pa
+        if integration_interval > pressure_hist.shape[0]:
+            integration_interval = pressure_hist.shape[0]
+
+        rms_p = np.sqrt(np.mean(np.square(pressure_hist[-integration_interval:-1]), axis=0))
+
+        rms_p[rms_p == 0.0] = 0.000000001
+        matrix_db = 20 * np.log10(rms_p / p0)
+        return matrix_db
+
+    def _evaluate(self, ind: Structure):
+        self.iteration = 0
+        obstacle_map = np.zeros((self.size_y, self.size_x))
+        pressure = np.zeros((self.size_y, self.size_x))
+        pressure_hist = np.zeros((self.duration, self.size_y, self.size_x))
+        velocities = np.zeros((self.size_y, self.size_x, 4))
+
+        for iteration in range(self.duration):
+            pressure_hist[iteration] = deepcopy(pressure)
+            velocities, pressure = self.step(velocities, pressure, obstacle_map)
+        # best_spl = self._reference_spl(sim)
+        spl = self.spl(pressure_hist)
+
+        current_spl = np.nan_to_num(spl, nan=0, neginf=0, posinf=0)
+        micro = Microphone(matrix=current_spl).array()
+        current_spl = np.concatenate(micro[1])
+        l_f = np.sum(np.abs(deepcopy(self.best_spl) - current_spl))
+        return l_f
+
+
+# # # Precompute domain arguments # # #
+
+pass
+
+# # #
+
+domain_cfg = Domain(
+    allowed_area=[
+        [0, 0],
+        [0, 120],
+        [120, 120],
+        [120, 0],
+        [0, 0],
+    ],
+    name='main',
+    min_poly_num=1,
+    max_poly_num=1,
+    min_points_num=3,
+    max_points_num=15,
+    polygon_side=0.0001,
+    min_dist_from_boundary=0.0001,
+    geometry_is_convex=True,
+    geometry_is_closed=True,
+    geometry='2D',
+)
+
+tuner_cfg = TunerParams(
+    tuner_type='optuna',
+    n_steps_tune=25,
+    hyperopt_dist='uniform',
+    verbose=True,
+    timeout_minutes=60,
+)
+
+
+best_structure = poly_from_comsol_txt(str(Path(__file__).parent) + '\\figures\\bottom_square.txt')
+best_spl = SoundSimulator(domain_cfg, 50, None)(best_structure)
+best_spl = np.nan_to_num(best_spl, nan=0, neginf=0, posinf=0)
+micro = Microphone(matrix=best_spl).array()
+best_spl = np.concatenate(micro[1])
+
+opt_params = OptimizationParams(
+    optimizer='gefest_ga',
+    domain=domain_cfg,
+    tuner_cfg=tuner_cfg,
+    n_steps=10,
+    pop_size=10,
+    postprocess_attempts=3,
+    mutation_prob=0.6,
+    crossover_prob=0.6,
+    mutations=[
+        'rotate_poly',
+        'resize_poly',
+        'add_point',
+        'drop_point',
+        'add_poly',
+        'drop_poly',
+        'pos_change_point',
+    ],
+    selector='tournament_selection',
+    mutation_each_prob=[0.125, 0.125, 0.15, 0.35, 0.00, 0.00, 0.25],
+    crossovers=[
+        'polygon_level',
+        'structure_level',
+    ],
+    crossover_each_prob=[0.0, 1.0],
+    postprocess_rules=[
+        'not_out_of_bounds',
+        'valid_polygon_geom',
+        'not_self_intersects',
+        'not_too_close_polygons',
+        # 'not_overlaps_prohibited',
+        'not_too_close_points',
+    ],
+    extra=5,
+    estimation_n_jobs=1,
+    n_jobs=-1,
+    log_dir='logs',
+    run_name='run_name',
+    golem_keep_histoy=False,
+    golem_genetic_scheme_type='steady_state',
+    golem_surrogate_each_n_gen=5,
+    objectives=[
+        SoundFieldFitness(
+            domain_cfg,
+            None,
+            best_spl,
+            50
+        ),
+    ],
+)

gefest/core/configs/optimization_params.py

@@ -77,7 +77,7 @@ class OptimizationParams(BaseModel):
     domain: Domain
     """Task domain."""
 
-    objectives: list[Objective]
+    objectives: list[Union[Objective, Callable]]
     """Task objectives."""
 
     mutations: Union[list[Callable[[Any], Structure]], list[ValidMutations]]

gefest/core/geometry/__init__.py

@@ -1,4 +1,4 @@
 from .datastructs.point import Point
 from .datastructs.polygon import Polygon, PolyID
 from .datastructs.structure import Structure
-from .utils import get_random_point, get_random_poly, get_random_structure
+from .utils import get_random_poly, get_random_structure

gefest/core/geometry/geometry_2d.py

@@ -38,7 +38,7 @@ class Geometry2D(Geometry):
 
     def get_length(self, polygon: Polygon):
         """Returns polygon perimeter."""
-        if len(polygon.points) < 1:
+        if len(polygon.points) <= 2:
             return 0
 
         geom_polygon = LineString([ShapelyPoint(pt.x, pt.y) for pt in polygon])
@@ -51,7 +51,7 @@ def shapely_to_gefest(self, geom_in):
             return Polygon(self.get_coords(geom_in))
         # add other shapely objects
 
-    def get_coords(self, poly) -> list[Point]:
+    def get_coords(self, poly: Union[ShapelyPolygon, LineString]) -> list[Point]:
         """The function for getting points.
 
         Args:
@@ -189,7 +189,7 @@ def get_square(self, polygon: Polygon) -> float:
         Returns:
             value of the :obj:`polygon` area.
         """
-        if len(polygon.points) <= 1:
+        if len(polygon.points) <= 2:
             return 0
 
         geom_polygon = ShapelyPolygon([self._pt_to_shapely_pt(pt) for pt in polygon])
@@ -331,7 +331,7 @@ def simplify(self, poly: Polygon, tolerance: float) -> Polygon:
         if len(poly) < 3:
             return poly
 
-        if self._poly_to_shapely_line(poly).is_simple:
+        if self._poly_to_shapely_poly(poly).is_simple:
 
             poly = self._poly_to_shapely_poly(inp)
             compressed = poly.buffer(-tolerance, join_style='mitre')
@@ -356,11 +356,9 @@ def simplify(self, poly: Polygon, tolerance: float) -> Polygon:
         else:
             simplified = self._poly_to_shapely_line(poly).convex_hull.simplify(tolerance)
             if simplified.is_empty:
-                raise ValueError('Empty polygon produced 2')
+                raise ValueError('Empty polygon produced')
 
             out = Polygon([Point(p[0], p[1]) for p in simplified.exterior.coords])
-        # plot_polygon(simplified, color='b')
-        # plt.show()
 
         return out
 

gefest/core/geometry/utils.py

@@ -86,21 +86,6 @@ def get_random_poly(parent_structure: Optional[Structure], domain: Domain) -> Op
     return polygon
 
 
-def get_random_point(polygon: Polygon, structure: Structure, domain: Domain) -> Optional[Point]:
-    """Generatres random point for polygon."""
-    centroid = domain.geometry.get_centroid(polygon)
-    sigma = _distance(centroid, structure, domain.geometry) / 3
-    point = create_polygon_point(centroid, sigma)
-    max_attempts = 20  # Number of attempts to create in bound point
-    while not _in_bound(point, domain):
-        point = create_polygon_point(centroid, sigma)
-        max_attempts -= 1
-        if max_attempts == 0:
-            return None
-
-    return point
-
-
 def create_poly(centroid: Point, sigma: int, domain: Domain, geometry: Geometry2D) -> Polygon:
     """Generates random polygon using poltgenerator lib.
 
@@ -191,45 +176,6 @@ def _create_area(domain: Domain, structure: Structure, geometry: Geometry2D) ->
     return centroid, sigma * 0.99
 
 
-def create_random_point(domain: Domain) -> Point:
-    """Returns random point from domain."""
-    point = Point(
-        np.random.uniform(low=domain.min_x, high=domain.max_x),
-        np.random.uniform(low=domain.min_y, high=domain.max_y),
-    )
-    while not _in_bound(point, domain):
-        point = Point(
-            np.random.uniform(low=domain.min_x, high=domain.max_x),
-            np.random.uniform(low=domain.min_y, high=domain.max_y),
-        )
-
-    return point
-
-
-def create_polygon_point(centroid: Point, sigma: int) -> Point:
-    """Generates point."""
-    point = Point(
-        np.random.normal(centroid.x, sigma, 1)[0],
-        np.random.normal(centroid.y, sigma, 1)[0],
-    )
-
-    return point
-
-
-def _in_bound(point: Point, domain: Domain) -> bool:
-    return domain.geometry.is_contain_point(domain.allowed_area, point)
-
-
-def _distance(point: Point, structure: Structure, geometry: Geometry2D) -> float:
-    polygons = structure.polygons
-    distances = []
-    for poly in polygons:
-        d = geometry.centroid_distance(point, poly)
-        distances.append(d)
-
-    return min(distances)
-
-
 def get_selfintersection_safe_point(
     poly: Polygon,
     domain: Domain,

gefest/tools/estimators/simulators/sound_wave/sound_interface.py

@@ -44,7 +44,7 @@ def generate_map(domain, structure):
     return obstacle_map
 
 
-def generate_random_map(map_size, random_seed):
+def generate_random_map(map_size: tuple[int, int], random_seed: int):
     """Randomly generate an array of zeros (free media) and ones (obstacles).
 
     The obstacles have basic geometric shapes.

gefest/tools/tuners/tuner.py

@@ -43,6 +43,9 @@ def __init__(
         opt_params: OptimizationParams,
         **kwargs,
     ) -> None:
+        if opt_params.tuner_cfg is None:
+            raise ValueError('TunerParams config not provided. Check cofiguration file.')
+
         self.log_dispatcher = opt_params.log_dispatcher
         self.domain: Domain = opt_params.domain
         self.validator = partial(validate, rules=opt_params.postprocess_rules, domain=self.domain)

gefest/tools/tuners/utils.py

@@ -4,31 +4,10 @@
 
 from gefest.core.geometry import Structure
 from gefest.core.geometry.domain import Domain
-from gefest.core.opt.objective.objective import Objective
 
 VarianceGeneratorType = Callable[[Structure], list[float]]
 
 
-def objective_validation_wrap(
-    struct: Structure,
-    objectives: Objective,
-    validator: Callable,
-) -> float:
-    """Applys validation rules to structure.
-
-    Used for GOLEM tuner as objective to filter out
-    invalid cases in tuning process.
-
-    Args:
-        struct (Structure): Changed stucture.
-        fitness_fun (Fitness): Fitness instance for current task.
-        validator (Callable): Util for structure validation.
-
-    Returns:
-        float: None if structure invalid else fintess function output
-    """
-
-
 def _get_uniform_args(mode: float, variance: float) -> tuple[float, float]:
     return (mode - variance, mode + variance)
 

setup.cfg

@@ -12,6 +12,7 @@ per-file-ignores =
     __init__.py:F401
     sound_interface.py:N806
     comsol_interface.py:C901
+    test\*:D103
 
 [tool:brunette]
 exclude = .git,.github,docs,__pycache__,env,venv,.venv