Loader refactor

.gitignore

@@ -168,3 +168,4 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 /.idea
+/.python-version

README.md

@@ -108,7 +108,7 @@ To use `brdr`, follow these steps:
 * Create a Aligner-class with specific parameters:
   * relevant_distance (m) (default: 1): Distance-parameter used to decide which parts will be aligned, and which parts remain unchanged.
   * od_strategy (enum) (default: SNAP_SINGLE_SIDE): Strategy to align geodata that is not covered by reference-data
-  * treshold_overlap_percentage (%)(0-100) (default 50)
+  * threshold_overlap_percentage (%)(0-100) (default 50)
   * crs: The Coordinate Reference System (CRS) (default: EPSG:31370 - Belgian Lambert72)
 * Load thematic data
 * Load reference data

brdr/__init__.py

@@ -6,4 +6,4 @@
     datefmt="%d-%b-%y %H:%M:%S",
 )
 
-__version__ = "0.1.0"
+__version__ = "0.1.1"

brdr/geometry_utils.py

@@ -2,12 +2,15 @@
 
 import numpy as np
 from shapely import GEOSException
+from shapely import GeometryCollection
 from shapely import Polygon
 from shapely import buffer
 from shapely import difference
 from shapely import intersection
 from shapely import is_empty
+from shapely import make_valid
 from shapely import symmetric_difference
+from shapely import unary_union
 from shapely import union
 from shapely.geometry.base import BaseGeometry
 
@@ -341,3 +344,26 @@ def grid_bounds(geom: BaseGeometry, delta: float):
             )
             grid.append(poly_ij)
     return grid
+
+
+def get_relevant_polygons_from_geom(geometry: BaseGeometry, buffer_distance: float):
+    """
+    Get only the relevant parts (polygon) from a geometry.
+    Points, Lines and Polygons smaller than relevant distance are excluded from the result
+    """
+    if not geometry or geometry.is_empty:
+        # If the input geometry is empty or None, do nothing.
+        return geometry
+    else:
+        geometry = make_valid(unary_union(geometry))
+        # Create a GeometryCollection from the input geometry.
+        geometry_collection = GeometryCollection(geometry)
+        array = []
+        for g in geometry_collection.geoms:
+            # Ensure each sub-geometry is valid.
+            g = make_valid(g)
+            if str(g.geom_type) in ["Polygon", "MultiPolygon"]:
+                relevant_geom = buffer_neg(g, buffer_distance)
+                if relevant_geom is not None and not relevant_geom.is_empty:
+                    array.append(g)
+    return make_valid(unary_union(array))

brdr/loader.py

@@ -0,0 +1,247 @@
+import json
+from abc import ABC
+
+import requests as requests
+from shapely import buffer
+from shapely import make_valid
+from shapely import unary_union
+from shapely.geometry import shape
+from shapely.geometry.base import BaseGeometry
+from shapely.prepared import prep
+
+from brdr.constants import *
+from brdr.enums import GRBType
+from brdr.geometry_utils import grid_bounds
+from brdr.typings import FeatureCollection
+from brdr.utils import get_collection
+
+
+class Loader(ABC):
+    def __init__(self):
+        self.data_dict: dict[str, BaseGeometry] = {}
+
+    def load_data(self):
+        return self.data_dict
+
+
+class DictLoader(Loader):
+    def __init__(self, data_dict: dict[str:BaseGeometry]):
+        super().__init__()
+        self.data_dict = data_dict
+
+    def load_data(self):
+        # self._prepare_reference_data()
+        return super().load_data()
+
+
+class GeoJsonLoader(Loader):
+    def __init__(
+        self,
+        _input: FeatureCollection,
+        id_property: str,
+    ):
+        super().__init__()
+        self.id_property = id_property
+        self.input = _input
+
+    def load_data(self):
+        self._load_geojson_data()
+        return super().load_data()
+
+    def _load_geojson_data(self):
+        """
+        Load geometries of a GeoJSON and stores them in a dictionary.
+
+        This method processes the thematic data from the input GeoJSON file. It
+        iterates through each feature, extracts the relevant properties, converts the
+        geometry to a valid shape, and stores it in a dictionary.
+
+        Returns:
+            None.
+        """
+        # THEMATIC PREPARATION
+        for f in self.input["features"]:
+            key = str(f["properties"][self.id_property])
+            geom = shape(f["geometry"])
+            self.data_dict[key] = make_valid(geom)
+        return
+
+
+class GeoJsonFileLoader(GeoJsonLoader):
+    def __init__(self, path_to_file, id_property):
+        with open(path_to_file, "r") as f:
+            _input = json.load(f)
+        super().__init__(_input, id_property)
+
+
+class GeoJsonUrlLoader(GeoJsonLoader):
+    def __init__(self, url, id_property):
+        _input = requests.get(url).json()
+        super().__init__(_input, id_property)
+
+
+class GRBActualLoader(Loader):
+    def __init__(self, grb_type: GRBType, partition: int, aligner):
+        super().__init__()
+        self.aligner = aligner
+        self.grb_type = grb_type
+        self.part = partition
+
+    def load_data(self):
+        if not self.aligner.dict_thematic:
+            raise ValueError("Thematic data not loaded")
+
+        self.load_reference_data_grb_actual(grb_type=self.grb_type, partition=self.part)
+        return super().load_data()
+
+    def load_reference_data_grb_actual(self, *, grb_type=GRBType.ADP, partition=0):
+        data_dict, id_property = self.get_reference_data_dict_grb_actual(
+            grb_type, partition
+        )
+        self.aligner.name_reference_id = id_property
+        self.aligner.logger.feedback_info(f"GRB downloaded: {grb_type}")
+
+        self.data_dict = data_dict
+
+    def get_reference_data_dict_grb_actual(self, grb_type=GRBType.ADP, partition=0):
+        """
+        Fetches reference data (administrative plots, buildings, or artwork) from the GRB
+        API based on thematic data.
+
+        This function retrieves reference data from the Grootschalig Referentie
+        Bestand (GRB) depending on the specified `grb_type` (e.g., administrative
+        plots (ADP), buildings (GBG), or artwork (KNW)).
+        It uses the bounding boxes of the geometries in the loaded thematic data
+        (`self.aligner.dict_thematic`) to filter the relevant reference data
+        geographically.
+
+        Args:
+            grb_type (GRBType, optional): The type of reference data to retrieve.
+                Defaults to GRBType.ADP (administrative plots).
+            partition (int, optional): If greater than zero, partitions the bounding box
+                of the thematic data into a grid before fetching reference data by
+                partition. Defaults to 0 (no partitioning).
+
+        Returns:
+            tuple: A tuple containing two elements:
+
+                - dict: A dictionary where keys are reference data identifiers
+                  (as defined by `name_reference_id`) and values are GeoJSON  geometry
+                  objects representing the reference data.
+                - str: The name of the reference data identifier property
+                  (e.g., "CAPAKEY" for ADP).
+
+        Raises:
+            ValueError: If an unsupported `grb_type` is provided.
+        """
+        if grb_type == GRBType.ADP:
+            url_grb = (
+                "https://geo.api.vlaanderen.be/GRB/ogc/features/collections/ADP/items?"
+            )
+            name_reference_id = "CAPAKEY"
+        elif grb_type == "gbg":
+            url_grb = (
+                "https://geo.api.vlaanderen.be/GRB/ogc/features/collections/GBG/items?"
+            )
+            name_reference_id = "OIDN"
+        elif grb_type == GRBType.KNW:
+            url_grb = (
+                "https://geo.api.vlaanderen.be/GRB/ogc/features/collections/KNW/items?"
+            )
+            name_reference_id = "OIDN"
+        else:
+            self.aligner.logger.feedback_info(
+                f"type not implemented: {str(grb_type)} -->No reference-data loaded"
+            )
+            return
+
+        crs = self.aligner.CRS
+        limit = DOWNLOAD_LIMIT
+        collection = {}
+        bounds_array = []
+
+        # Get the bounds of the thematic_data to get the necessary GRB-data
+        for key in self.aligner.dict_thematic:
+            # buffer them geometry with x m (default 10)
+            buffer_value = self.aligner.relevant_distance + MAX_REFERENCE_BUFFER
+            geom = buffer(
+                self.aligner.dict_thematic[key],
+                buffer_value,
+                quad_segs=QUAD_SEGMENTS,
+                join_style="mitre",
+                mitre_limit=MITRE_LIMIT,
+            )
+            bounds_array.append(geom)
+            if partition < 1:
+                bbox = str(geom.bounds).strip("()")
+                url_grb_bbox = (
+                    url_grb
+                    + "f=application%2Fgeo%2Bjson&limit="
+                    + str(limit)
+                    + "&crs="
+                    + crs
+                    + "&bbox-crs="
+                    + crs
+                    + "&bbox="
+                    + bbox
+                )
+                self.aligner.logger.feedback_debug(key + "-->" + str(url_grb_bbox))
+                coll = self._get_dict_from_url(url_grb_bbox, name_reference_id, limit)
+                collection.update(coll)
+            if partition > 0:
+                geom = unary_union(bounds_array)
+                grid = self.partition(geom, partition)
+                for g in grid:
+                    bbox = str(g.bounds).strip("()")
+                    url_grb_bbox = (
+                        url_grb
+                        + "f=application%2Fgeo%2Bjson&limit="
+                        + str(limit)
+                        + "&crs="
+                        + crs
+                        + "&bbox-crs="
+                        + crs
+                        + "&bbox="
+                        + bbox
+                    )
+                    self.aligner.logger.feedback_debug(key + "-->" + str(url_grb_bbox))
+                    coll = self._get_dict_from_url(
+                        url_grb_bbox, name_reference_id, limit
+                    )
+                    collection.update(coll)
+
+        return collection, name_reference_id
+
+    @staticmethod
+    def partition(geom, delta):
+        """
+        Filters a computed grid of partitions (generated by `_grid_bounds`) based on
+        intersection with a geometric object (`geom`).
+
+        Args:
+            geom (BaseGeometry): The geometric object to check for intersection
+                with partitions.
+            delta (float): The distance between partitions (same value used in
+                `_grid_bounds`).
+
+        Returns:
+            list: A filtered list of Polygon objects representing the partitions
+                overlapping the original geometric object.
+        """
+        prepared_geom = prep(geom)
+        partitions = grid_bounds(geom, delta)
+        filtered_grid = list(filter(prepared_geom.intersects, partitions))
+        return filtered_grid
+
+    def _get_dict_from_url(self, input_url, name_reference_id, limit):
+        collection = get_collection(input_url, limit)
+        dictionary = {}
+        if "features" not in collection or len(collection["features"]) == 0:
+            return dictionary
+        for f in collection["features"]:
+            key = str(f["properties"][name_reference_id])
+            geom = shape(f["geometry"])
+            if key not in collection:
+                dictionary[key] = make_valid(geom)
+            self.aligner.logger.feedback_debug(key + "-->" + str(geom))
+        return dictionary

brdr/logger.py

@@ -0,0 +1,28 @@
+import logging
+
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(message)s", datefmt="%d-%b-%y %H:%M:%S"
+)
+
+
+class Logger:
+    def __init__(self, feedback=None):
+        self.feedback = feedback
+
+    def feedback_debug(self, text):
+        if self.feedback is not None:
+            # self.feedback.pushInfo(text)
+            return
+        logging.debug(text)
+
+    def feedback_info(self, text):
+        if self.feedback is not None:
+            self.feedback.pushInfo(text)
+            return
+        logging.info(text)
+
+    def feedback_warning(self, text):
+        if self.feedback is not None:
+            self.feedback.pushInfo(text)
+            return
+        logging.warning(text)

brdr/typings.py

@@ -0,0 +1,28 @@
+# define a typeddict thematic_data with keys name: str and geom: geometry
+from typing import Dict
+from typing import List
+from typing import TypedDict
+
+
+class GeoJSONGeometry(TypedDict):
+    type: str
+    coordinates: List
+
+
+class Crs(TypedDict):
+    type: str
+    properties: dict
+
+
+class Feature(TypedDict):
+    type: str
+    geometry: GeoJSONGeometry
+    properties: dict
+
+
+class FeatureCollection(TypedDict, total=False):
+    type: str
+    name: str
+    crs: Crs
+    features: List[Feature]
+    __extra_items__: Dict[str, str]

brdr/utils.py

@@ -91,28 +91,6 @@ def geojson_tuple_from_dict_theme(
     return tuple(feature_collections)
 
 
-def geojson_tuple_from_dict_theme(
-    dict_theme, crs, name_id, prop_dict=None, geom_attributes=True
-):
-    """
-    get a geojson-tuple (6 geojsons) for a dictionary of theme_ids (keys) and dictionary of relevant distance-results (values)
-    """
-    features = [[], [], [], [], [], []]
-    for key in dict_theme.keys():
-        if prop_dict is not None and key in prop_dict:
-            prop_dictionary = prop_dict[key]
-        fcs = geojson_tuple_from_series(
-            dict_theme[key], crs, name_id, prop_dict=prop_dictionary
-        )
-        for count, ft in enumerate(features):
-            ft.extend(fcs[count].features)
-    crs_geojson = {"type": "name", "properties": {"name": crs}}
-    feature_collections = []
-    for ft in features:
-        feature_collections.append(FeatureCollection(ft, crs=crs_geojson))
-    return tuple(feature_collections)
-
-
 def geojson_from_dict(dictionary, crs, name_id, prop_dict=None, geom_attributes=True):
     """
     get a geojson (featurecollection) from a dictionary of ids(keys) and geometries (values)