diff --git a/geest/core/algorithms/ors_multibuffer_processor.py b/geest/core/algorithms/ors_multibuffer_processor.py index d6d69c17..0588386c 100644 --- a/geest/core/algorithms/ors_multibuffer_processor.py +++ b/geest/core/algorithms/ors_multibuffer_processor.py @@ -742,8 +742,8 @@ def rasterize( ) # Ensure resolution parameters are properly formatted as float values - x_res = 100.0 # 100m pixel size in X direction - y_res = 100.0 # 100m pixel size in Y direction + x_res = self.cell_size_m # pixel size in X direction + y_res = self.cell_size_m # pixel size in Y direction bbox = bbox.boundingBox() # Define rasterization parameters for the temporary layer params = { @@ -755,7 +755,6 @@ def rasterize( "WIDTH": x_res, "HEIGHT": y_res, "EXTENT": f"{bbox.xMinimum()},{bbox.xMaximum()},{bbox.yMinimum()},{bbox.yMaximum()} [{self.target_crs.authid()}]", - #'EXTENT':'280518.114000000,296308.326900000,3998456.316800000,4003763.812500000 [EPSG:32630]', "NODATA": 0, "OPTIONS": "", "DATA_TYPE": 0, diff --git a/geest/core/algorithms/safety_polygon_processor.py b/geest/core/algorithms/safety_polygon_processor.py index b13b65a5..a72e764e 100644 --- a/geest/core/algorithms/safety_polygon_processor.py +++ b/geest/core/algorithms/safety_polygon_processor.py @@ -33,11 +33,23 @@ class SafetyPerCellProcessor: - Assigning values based on perceived safety. - Rasterizing the results. - Combining rasterized results into a single VRT file. + + Args: + output_prefix (str): Prefix for the output files. + cell_size_m (float): The cell size in meters for the analysis. + safety_layer (QgsVectorLayer): The input layer containing safety data. + safety_field (str): The field in the safety layer containing perceived safety values. + workflow_directory (str): Directory where output files will be stored. + gpkg_path (str): Path to the GeoPackage with study areas. + context (QgsProcessingContext): The processing context to pass objects to the thread. + Returns: + str: The path to the created VRT file. """ def __init__( self, output_prefix: str, + cell_size_m: float, safety_layer: QgsVectorLayer, safety_field: str, workflow_directory: str, @@ -54,6 +66,7 @@ def __init__( gpkg_path (str): Path to the GeoPackage with study areas. """ self.output_prefix = output_prefix + self.cell_size_m = cell_size_m self.safety_layer = safety_layer self.workflow_directory = workflow_directory self.gpkg_path = gpkg_path @@ -187,8 +200,8 @@ def _rasterize_safety( "BURN": None, "USE_Z": False, "UNITS": 1, - "WIDTH": 100.0, - "HEIGHT": 100.0, + "WIDTH": self.cell_size_m, + "HEIGHT": self.cell_size_m, "EXTENT": f"{bbox.xMinimum()},{bbox.xMaximum()},{bbox.yMinimum()},{bbox.yMaximum()}", "NODATA": 255, "DATA_TYPE": 0, diff --git a/geest/core/algorithms/single_point_buffer_processor.py b/geest/core/algorithms/single_point_buffer_processor.py index 9de00f60..bde7b6fd 100644 --- a/geest/core/algorithms/single_point_buffer_processor.py +++ b/geest/core/algorithms/single_point_buffer_processor.py @@ -24,6 +24,7 @@ class SinglePointBufferProcessor: def __init__( self, output_prefix: str, + cell_size_m: float, input_layer: QgsVectorLayer, buffer_distance: float, workflow_directory: str, @@ -34,12 +35,14 @@ def __init__( Args: output_prefix (str): Prefix for naming output files. + cell_size_m (float): The cell size in meters for the analysis. input_layer (QgsVectorLayer): The input layer containing the points to buffer. buffer_distance (float): The distance in meters to buffer the points. workflow_directory (str): Directory where temporary and output files will be stored. gpkg_path (str): Path to the GeoPackage containing study areas and bounding boxes. """ self.output_prefix = output_prefix + self.cell_size_m = cell_size_m self.features_layer = input_layer self.buffer_distance = buffer_distance self.workflow_directory = workflow_directory @@ -287,8 +290,8 @@ def _rasterize( QgsMessageLog.logMessage(f"Rasterizing {input_layer}", "Geest", Qgis.Info) # Ensure resolution parameters are properly formatted as float values - x_res = 100.0 # 100m pixel size in X direction - y_res = 100.0 # 100m pixel size in Y direction + x_res = self.cell_size_m # pixel size in X direction + y_res = self.cell_size_m # pixel size in Y direction bbox = bbox.boundingBox() # Define rasterization parameters for the temporary layer params = { @@ -386,8 +389,8 @@ def _mask_raster( "CROP_TO_CUTLINE": True, "KEEP_RESOLUTION": False, "SET_RESOLUTION": True, - "X_RESOLUTION": 100, - "Y_RESOLUTION": 100, + "X_RESOLUTION": self.cell_size_m, + "Y_RESOLUTION": self.cell_size_m, "MULTITHREADING": True, "DATA_TYPE": 0, # byte "EXTRA": "", diff --git a/geest/core/buffering.py b/geest/core/buffering.py deleted file mode 100644 index 8bbbc072..00000000 --- a/geest/core/buffering.py +++ /dev/null @@ -1,96 +0,0 @@ -import os -from qgis.core import ( - QgsVectorLayer, - QgsProcessingFeedback, - QgsMessageLog, - Qgis, -) -import processing - - -class SinglePointBuffer: - def __init__(self, input_layer, buffer_distance, output_path, crs): - """ - Initializes the SinglePointBuffer class. - - Args: - input_layer (QgsVectorLayer): The input polygon or line layer to buffer. - buffer_distance (float): The distance of the buffer in CRS units. - output_path (str): The path to save the buffered layer. - crs (QgsCoordinateReferenceSystem): The expected CRS for the input layer. - """ - self.input_layer = input_layer - self.buffer_distance = buffer_distance - self.output_path = output_path - self.crs = crs - - # Check if the input layer CRS matches the expected CRS, and reproject if necessary - self.processed_layer = self._check_and_reproject_layer() - - def _check_and_reproject_layer(self): - """ - Checks if the input layer has the expected CRS. If not, it reprojects the layer. - - Returns: - QgsVectorLayer: The input layer, either reprojected or unchanged. - """ - if self.input_layer.crs() != self.crs: - QgsMessageLog.logMessage( - f"Reprojecting layer from {self.input_layer.crs().authid()} to {self.crs.authid()}", - "Geest", - level=Qgis.Info, - ) - reproject_result = processing.run( - "native:reprojectlayer", - { - "INPUT": self.input_layer, - "TARGET_CRS": self.crs, - "OUTPUT": "memory:", # Reproject in memory - }, - feedback=QgsProcessingFeedback(), - ) - reprojected_layer = reproject_result["OUTPUT"] - return reprojected_layer - - # If CRS matches, return the original layer - return self.input_layer - - def create_buffer(self): - """ - Creates a buffer around the input layer's geometries. - - Returns: - QgsVectorLayer: The resulting buffered layer. - """ - # Check if the output file already exists and delete it if necessary - if os.path.exists(self.output_path): - QgsMessageLog.logMessage( - f"Warning: {self.output_path} already exists. It will be overwritten.", - "Geest", - level=Qgis.Warning, - ) - os.remove(self.output_path) - - # Run the buffer operation using QGIS processing - buffer_result = processing.run( - "native:buffer", - { - "INPUT": self.processed_layer, - "DISTANCE": self.buffer_distance, - "SEGMENTS": 5, # The number of segments used to approximate curves - "END_CAP_STYLE": 0, # Round cap - "JOIN_STYLE": 0, # Round joins - "MITER_LIMIT": 2, - "DISSOLVE": False, # Whether to dissolve the output or keep it separate for each feature - "OUTPUT": self.output_path, - }, - feedback=QgsProcessingFeedback(), - ) - - # Load the buffered layer as QgsVectorLayer - buffered_layer = QgsVectorLayer(self.output_path, "buffered_layer", "ogr") - - if not buffered_layer.isValid(): - raise ValueError("Buffered layer creation failed.") - - return buffered_layer diff --git a/geest/core/convert_to_8bit.py b/geest/core/convert_to_8bit.py deleted file mode 100644 index 6c6bf154..00000000 --- a/geest/core/convert_to_8bit.py +++ /dev/null @@ -1,57 +0,0 @@ -import os -import processing -from qgis.core import QgsMessageLog, Qgis, QgsRasterLayer, QgsProject - - -class RasterConverter: - """ - A class to handle the conversion of rasters to 8-bit TIFFs. - """ - - def __init__(self, feedback=None): - """ - Initialize the RasterConverter with optional feedback for progress reporting. - :param feedback: Optional QgsFeedback object for reporting progress. - """ - self.feedback = feedback - - def convert_to_8bit(self, input_raster: str, output_raster: str) -> bool: - """ - Convert the input raster to an 8-bit TIFF using gdal:translate. - :param input_raster: Path to the input raster file. - :param output_raster: Path to the output 8-bit TIFF file. - :return: True if conversion is successful, False otherwise. - """ - QgsMessageLog.logMessage( - f"Converting {input_raster} to 8-bit TIFF at {output_raster}.", - tag="Geest", - level=Qgis.Info, - ) - - params = { - "INPUT": input_raster, - "TARGET_CRS": None, # Use input CRS - "NODATA": -9999, - "COPY_SUBDATASETS": False, - "OPTIONS": "", - "EXTRA": "", - "DATA_TYPE": 1, # 1 = Byte (8-bit unsigned) - "OUTPUT": output_raster, - } - - try: - # Run the gdal:translate processing algorithm - processing.run("gdal:translate", params, feedback=self.feedback) - QgsMessageLog.logMessage( - f"Successfully converted {input_raster} to 8-bit TIFF.", - tag="Geest", - level=Qgis.Info, - ) - return True - except Exception as e: - QgsMessageLog.logMessage( - f"Failed to convert {input_raster} to 8-bit: {str(e)}", - tag="Geest", - level=Qgis.Critical, - ) - return False diff --git a/geest/core/create_grids.py b/geest/core/create_grids.py deleted file mode 100644 index 52eabbb2..00000000 --- a/geest/core/create_grids.py +++ /dev/null @@ -1,147 +0,0 @@ -import os -from qgis.core import QgsVectorLayer, QgsProcessingFeedback, QgsMessageLog, Qgis -import processing - - -class GridCreator: - def __init__(self, h_spacing=100, v_spacing=100): - """ - Initializes the GridCreator class with default grid spacing. - - Args: - h_spacing (float): Horizontal spacing for the grid (default is 100 meters). - v_spacing (float): Vertical spacing for the grid (default is 100 meters). - """ - self.h_spacing = h_spacing - self.v_spacing = v_spacing - - def create_grids(self, layer, output_dir, crs, merged_output_path): - """ - Creates grids for the input polygon layer, clips them, and merges them into a single grid. - - Args: - layer (QgsVectorLayer): The input polygon layer to grid. - output_dir (str): The directory to save the grid outputs. - crs (QgsCoordinateReferenceSystem): The coordinate reference system (CRS) for the grids. - merged_output_path (str): The output path for the merged grid. - - Returns: - QgsVectorLayer: The merged grid layer. - """ - # Check if the merged grid already exists - if os.path.exists(merged_output_path): - QgsMessageLog.logMessage( - f"Merged grid already exists: {merged_output_path}", - "Geest", - level=Qgis.Info, - ) - return merged_output_path - else: - layer = QgsVectorLayer(layer, "country_layer", "ogr") - if not layer.isValid(): - raise ValueError("Invalid country layer") - - # Reproject the country layer if necessary - if layer.crs() != crs: - layer = processing.run( - "native:reprojectlayer", - { - "INPUT": layer, - "TARGET_CRS": crs, - "OUTPUT": "memory:", - }, - feedback=QgsProcessingFeedback(), - )["OUTPUT"] - - all_grids = [] - - # Loop through each feature in the polygon layer - for feature in layer.getFeatures(): - geom = feature.geometry() - - # Check if the geometry is multipart - if geom.isMultipart(): - parts = ( - geom.asGeometryCollection() - ) # Separate multipart geometry into parts - else: - parts = [geom] # Single part geometry - - # Loop through each part of the geometry - for part_id, part in enumerate(parts): - part_area = part.area() - - # Get the extent of each part - part_extent = part.boundingBox() - - # Define the output grid path for each part - grid_output_path = ( - f"{output_dir}/grid_{feature.id()}_part_{part_id}.shp" - ) - - # Check if the grid already exists - if os.path.exists(grid_output_path): - QgsMessageLog.logMessage( - f"Grid file already exists: {grid_output_path}", - "Geest", - level=Qgis.Info, - ) - grid_layer = QgsVectorLayer( - grid_output_path, "grid_layer", "ogr" - ) # Load the existing grid layer - # Clip the grid to the polygon feature (to restrict it to the boundaries) - clipped_grid_output_path = f"{output_dir}/clipped_grid_{feature.id()}_part_{part_id}.shp" - clip_params = { - "INPUT": grid_layer, # The grid we just created - "OVERLAY": layer, # The layer we're clipping to - "OUTPUT": clipped_grid_output_path, - } - clip_result = processing.run("native:clip", clip_params) - grid_layer = clip_result["OUTPUT"] # The clipped grid - else: - QgsMessageLog.logMessage( - f"Creating grid: {grid_output_path}", - "Geest", - level=Qgis.Info, - ) - # Define grid creation parameters - grid_params = { - "TYPE": 2, # Rectangle (polygon) - "EXTENT": part_extent, # Use the extent of the current part - "HSPACING": self.h_spacing, # Horizontal spacing - "VSPACING": self.v_spacing, # Vertical spacing - "CRS": crs, # Coordinate reference system (CRS) - "OUTPUT": grid_output_path, # Output path for the grid file - } - - # Create the grid using QGIS processing - grid_result = processing.run("native:creategrid", grid_params) - grid_layer = grid_result["OUTPUT"] # Get the grid layer - - # Clip the grid to the polygon feature (to restrict it to the boundaries) - clipped_grid_output_path = f"{output_dir}/clipped_grid_{feature.id()}_part_{part_id}.shp" - clip_params = { - "INPUT": grid_layer, # The grid we just created - "OVERLAY": layer, # The layer we're clipping to - "OUTPUT": clipped_grid_output_path, - } - clip_result = processing.run("native:clip", clip_params) - grid_layer = clip_result["OUTPUT"] # The clipped grid - - # Add the generated or loaded grid to the list - all_grids.append(grid_layer) - - # Merge all grids into a single layer - QgsMessageLog.logMessage( - f"Merging grids into: {merged_output_path}", "Geest", level=Qgis.Info - ) - merge_params = { - "LAYERS": all_grids, - "CRS": crs, - "OUTPUT": merged_output_path, - } - merged_grid = processing.run("native:mergevectorlayers", merge_params)[ - "OUTPUT" - ] - - return merged_grid diff --git a/geest/core/crs_converter.py b/geest/core/crs_converter.py deleted file mode 100644 index 49c4ac06..00000000 --- a/geest/core/crs_converter.py +++ /dev/null @@ -1,60 +0,0 @@ -from qgis.core import ( - QgsCoordinateReferenceSystem, - QgsProcessingFeedback, - QgsMessageLog, - Qgis, -) -from qgis import processing - - -class CRSConverter: - def __init__(self, layer): - """ - Initialize the CRSConverter class with a given layer. - :param layer: The input layer for CRS conversion (QgsVectorLayer or QgsRasterLayer) - """ - self.layer = layer - - def convert_to_crs(self, target_crs_epsg): - """ - Converts the layer's CRS to the target CRS based on the EPSG code. - :param target_crs_epsg: EPSG code of the target CRS - """ - # Get the current CRS of the layer - current_crs = self.layer.crs() - - # Create the target CRS using the EPSG code - target_crs = QgsCoordinateReferenceSystem(f"EPSG:{target_crs_epsg}") - - # Check if the current CRS is the same as the target CRS - if current_crs != target_crs: - QgsMessageLog.logMessage( - f"Converting layer from {current_crs.authid()} to {target_crs.authid()}", - tag="Geest", - level=Qgis.Info, - ) - - layer = processing.run( - "native:reprojectlayer", - { - "INPUT": self.layer, - "TARGET_CRS": target_crs, - "OUTPUT": "memory:", - }, - feedback=QgsProcessingFeedback(), - )["OUTPUT"] - QgsMessageLog.logMessage( - f"Layer successfully converted to {target_crs.authid()}", - tag="Geest", - level=Qgis.Info, - ) - - return layer - else: - QgsMessageLog.logMessage( - f"Layer is already in the target CRS: {target_crs.authid()}", - tag="Geest", - level=Qgis.Info, - ) - - return self.layer diff --git a/geest/core/extents.py b/geest/core/extents.py deleted file mode 100644 index 358695cf..00000000 --- a/geest/core/extents.py +++ /dev/null @@ -1,89 +0,0 @@ -import os -from qgis.core import QgsVectorLayer, QgsProcessingFeedback -import processing - - -class Extents: - def __init__(self, workingDir, countryLayerPath, pixelSize, UTM_crs): - """ - Initializes the Extents class with relevant parameters. - - Args: - workingDir (str): The working directory path. - countryLayerPath (str): The file path of the country layer. - pixelSize (float): The pixel size for raster operations. - UTM_crs (QgsCoordinateReferenceSystem): The CRS to reproject the country layer to. - """ - # Set up paths and directories - self.current_script_path = os.path.dirname(os.path.abspath(__file__)) - self.workingDir = os.path.normpath(workingDir) - self.Dimension = "Place Characterization" - self.tempDir = os.path.join(self.workingDir, "temp") - - # Create necessary directories - self._setup_directories() - - # Input parameters - self.countryLayerPath = countryLayerPath - self.pixelSize = pixelSize - self.UTM_crs = UTM_crs - - # Preprocess the country layer - self._load_and_preprocess_country_layer() - - def _setup_directories(self): - """Sets up the working and temporary directories.""" - os.makedirs(os.path.join(self.workingDir, self.Dimension), exist_ok=True) - os.makedirs(self.tempDir, exist_ok=True) - os.chdir(self.workingDir) - - def _load_and_preprocess_country_layer(self): - """Loads and preprocesses the country layer, including reprojecting if necessary.""" - # Load the country layer - self.countryLayer = QgsVectorLayer( - self.countryLayerPath, "country_layer", "ogr" - ) - if not self.countryLayer.isValid(): - raise ValueError("Invalid country layer") - - # Reproject the country layer if necessary - if self.countryLayer.crs() != self.UTM_crs: - self.countryLayer = processing.run( - "native:reprojectlayer", - { - "INPUT": self.countryLayer, - "TARGET_CRS": self.UTM_crs, - "OUTPUT": "memory:", - }, - feedback=QgsProcessingFeedback(), - )["OUTPUT"] - - # Get the extent of the country layer - self.country_extent = self.countryLayer.extent() - - def get_country_extent(self): - """ - Returns the extent of the country layer. - - Returns: - QgsRectangle: Extent of the country layer. - """ - return self.country_extent - - def get_processed_layers(self): - """ - Returns a dictionary containing the processed layers and paths. - - Returns: - dict: Contains processed layers and paths. - """ - return { - "current_script_path": self.current_script_path, - "workingDir": self.workingDir, - "Dimension": self.Dimension, - "tempDir": self.tempDir, - "countryLayer": self.countryLayer, - "country_extent": self.country_extent, - "pixelSize": self.pixelSize, - "UTM_crs": self.UTM_crs, - } diff --git a/geest/core/json_tree_item.py b/geest/core/json_tree_item.py index 01e2ba6e..daa1fa5c 100644 --- a/geest/core/json_tree_item.py +++ b/geest/core/json_tree_item.py @@ -1,9 +1,8 @@ import uuid -from qgis.PyQt.QtCore import Qt +import traceback -# Change to this when implementing in QGIS -# from qgis.PyQt.QtGui import ( -from PyQt5.QtGui import QColor, QFont, QIcon +from qgis.PyQt.QtCore import Qt +from qgis.PyQt.QtGui import QColor, QFont, QIcon from qgis.core import QgsMessageLog, Qgis from geest.utilities import resources_path from geest.core import setting @@ -168,6 +167,11 @@ def getStatusTooltip(self): def getStatus(self): """Return the status of the item as single character.""" try: + if not type(self.itemData) == list: + return "" + if len(self.itemData) < 4: + return "" + data = self.itemData[3] # QgsMessageLog.logMessage(f"Data: {data}", tag="Geest", level=Qgis.Info) status = "" @@ -176,12 +180,12 @@ def getStatus(self): if "Failed" in data.get("result", ""): return "x" # Item required and not configured - if "Don’t Use" in data.get("analysis_mode", "") and data.get( + if "Do Not Use" in data.get("analysis_mode", "") and data.get( "indicator_required", False ): return "-" # Item not required but not configured - if "Don’t Use" in data.get("analysis_mode", "") and not data.get( + if "Do Not Use" in data.get("analysis_mode", "") and not data.get( "indicator_required", False ): return "!" @@ -191,10 +195,8 @@ def getStatus(self): return "✔️" except Exception as e: - verbose_mode = setting.value("verbose_mode", False) + verbose_mode = setting("verbose_mode", False) if verbose_mode: - import traceback - QgsMessageLog.logMessage( f"Error getting status: {e}", tag="Geest", level=Qgis.Warning ) @@ -222,6 +224,9 @@ def updateStatus(self, status=None): QgsMessageLog.logMessage( f"Error updating status: {e}", tag="Geest", level=Qgis.Warning ) + QgsMessageLog.logMessage( + traceback.format_exc(), tag="Geest", level=Qgis.Warning + ) def getFont(self): """Retrieve the appropriate font for the item based on its role.""" @@ -320,6 +325,7 @@ def getAnalysisAttributes(self): "analysis_description", "Not Set" ) attributes["working_folder"] = self.data(3).get("working_folder", "Not Set") + attributes["cell_size_m"] = self.data(3).get("cell_size_m", 100) attributes["dimensions"] = [ { diff --git a/geest/core/points_per_grid_cell.py b/geest/core/points_per_grid_cell.py deleted file mode 100644 index a1dd6b53..00000000 --- a/geest/core/points_per_grid_cell.py +++ /dev/null @@ -1,139 +0,0 @@ -import os -from qgis.PyQt.QtCore import QVariant -from qgis.core import ( - QgsVectorLayer, - QgsField, - QgsSpatialIndex, - QgsProcessingFeedback, -) -import processing -from .create_grids import GridCreator -from .extents import Extents - - -class RasterPointGridScore: - def __init__(self, country_boundary, pixel_size, output_path, crs, input_points): - self.country_boundary = country_boundary - self.pixel_size = pixel_size - self.output_path = output_path - self.crs = crs - self.input_points = input_points - - def raster_point_grid_score(self): - """ - Generates a raster based on the number of input points within each grid cell. - :param country_boundary: Layer defining the country boundary to clip the grid. - :param cellsize: The size of each grid cell. - :param output_path: Path to save the output raster. - :param crs: The CRS in which the grid and raster will be projected. - :param input_points: Layer of point features to count within each grid cell. - """ - - # Create grid - self.h_spacing = 100 - self.v_spacing = 100 - create_grid = GridCreator(h_spacing=self.h_spacing, v_spacing=self.v_spacing) - output_dir = os.path.join("output") - merged_output_path = os.path.join( - output_dir, "merged_grid.shp" - ) # Use Shapefile - - # Create grid layer using Shapefile - grid_layer = create_grid.create_grids( - self.country_boundary, output_dir, self.crs, merged_output_path - ) - grid_layer = QgsVectorLayer(merged_output_path, "merged_grid", "ogr") - - # Add score field - provider = grid_layer.dataProvider() - field_name = "score" - if not grid_layer.fields().indexFromName(field_name) >= 0: - provider.addAttributes([QgsField(field_name, QVariant.Int)]) - grid_layer.updateFields() - - # Create spatial index for the input points - # Reproject the country layer if necessary - if self.input_points.crs() != self.crs: - self.input_points = processing.run( - "native:reprojectlayer", - { - "INPUT": self.input_points, - "TARGET_CRS": self.crs, - "OUTPUT": "memory:", - }, - feedback=QgsProcessingFeedback(), - )["OUTPUT"] - point_index = QgsSpatialIndex(self.input_points.getFeatures()) - - # Count points within each grid cell and assign a score - reclass_vals = {} - for grid_feat in grid_layer.getFeatures(): - grid_geom = grid_feat.geometry() - # Get intersecting points - intersecting_points = point_index.intersects(grid_geom.boundingBox()) - num_points = len(intersecting_points) - - # Reclassification logic: assign score based on the number of points - if num_points >= 2: - reclass_val = 5 - elif num_points == 1: - reclass_val = 3 - else: - reclass_val = 0 - - reclass_vals[grid_feat.id()] = reclass_val - - # Apply the score values to the grid - grid_layer.startEditing() - for grid_feat in grid_layer.getFeatures(): - grid_layer.changeAttributeValue( - grid_feat.id(), - provider.fieldNameIndex(field_name), - reclass_vals[grid_feat.id()], - ) - grid_layer.commitChanges() - - merged_output_vector = os.path.join( - output_dir, "merged_grid_vector.shp" - ) # Use Shapefile for merged output - - # Merge grids into a single Shapefile layer - Merge = processing.run( - "native:mergevectorlayers", - {"LAYERS": [grid_layer], "CRS": None, "OUTPUT": "memory:"}, - ) - - merge = Merge["OUTPUT"] - - extents_processor = Extents( - output_dir, self.country_boundary, self.pixel_size, self.crs - ) - - # Get the extent of the vector layer - country_extent = extents_processor.get_country_extent() - xmin, ymin, xmax, ymax = ( - country_extent.xMinimum(), - country_extent.yMinimum(), - country_extent.xMaximum(), - country_extent.yMaximum(), - ) - - # Rasterize the clipped grid layer to generate the raster - rasterize_params = { - "INPUT": merge, - "FIELD": field_name, - "BURN": 0, - "USE_Z": False, - "UNITS": 1, - "WIDTH": self.pixel_size, - "HEIGHT": self.pixel_size, - "EXTENT": f"{xmin},{xmax},{ymin},{ymax}", - "NODATA": -9999, - "OPTIONS": "", - "DATA_TYPE": 5, # Use Int32 for scores - "OUTPUT": self.output_path, - } - - processing.run( - "gdal:rasterize", rasterize_params, feedback=QgsProcessingFeedback() - ) diff --git a/geest/core/polygons_per_grid_cell.py b/geest/core/polygons_per_grid_cell.py deleted file mode 100644 index c9c3d185..00000000 --- a/geest/core/polygons_per_grid_cell.py +++ /dev/null @@ -1,208 +0,0 @@ -import os -from qgis.PyQt.QtCore import QVariant -from qgis.core import ( - QgsGeometry, - QgsVectorLayer, - QgsField, - QgsSpatialIndex, - QgsProcessingFeedback, -) -import processing -from .utilities import GridAligner - - -class RasterPolygonGridScore: - def __init__( - self, - country_boundary, - pixel_size, - working_dir, - crs, - input_polygons, - output_path, - ): - self.country_boundary = country_boundary - self.pixel_size = pixel_size - self.working_dir = working_dir - self.crs = crs - self.input_polygons = input_polygons - self.output_path = output_path - # Initialize GridAligner with grid size - self.grid_aligner = GridAligner(grid_size=100) - - def raster_polygon_grid_score(self): - """ - Generates a raster based on the number of input points within each grid cell. - :param country_boundary: Layer defining the country boundary to clip the grid. - :param cellsize: The size of each grid cell. - :param output_path: Path to save the output raster. - :param crs: The CRS in which the grid and raster will be projected. - :param input_polygons: Layer of point features to count within each grid cell. - """ - - output_dir = os.path.dirname(self.output_path) - - # Define output directory and ensure it's created - os.makedirs(output_dir, exist_ok=True) - - # Load grid layer from the Geopackage - geopackage_path = os.path.join( - self.working_dir, "study_area", "study_area.gpkg" - ) - if not os.path.exists(geopackage_path): - raise ValueError(f"Geopackage not found at {geopackage_path}.") - - grid_layer = QgsVectorLayer( - f"{geopackage_path}|layername=study_area_grid", "merged_grid", "ogr" - ) - - area_layer = QgsVectorLayer( - f"{geopackage_path}|layername=study_area_polygons", - "study_area_polygons", - "ogr", - ) - - geometries = [feature.geometry() for feature in area_layer.getFeatures()] - - # Combine all geometries into one using unaryUnion - area_geometry = QgsGeometry.unaryUnion(geometries) - - # grid_geometry = grid_layer.getGeometry() - - aligned_bbox = self.grid_aligner.align_bbox( - area_geometry.boundingBox(), area_layer.extent() - ) - - # Extract polylines by location - grid_output = processing.run( - "native:extractbylocation", - { - "INPUT": grid_layer, - "PREDICATE": [0], - "INTERSECT": self.input_polygons, - "OUTPUT": "TEMPORARY_OUTPUT", - }, - feedback=QgsProcessingFeedback(), - )["OUTPUT"] - - grid_layer = grid_output - - # Add score field - provider = grid_layer.dataProvider() - field_name = "poly_score" - if not grid_layer.fields().indexFromName(field_name) >= 0: - provider.addAttributes([QgsField(field_name, QVariant.Int)]) - grid_layer.updateFields() - - # Create spatial index for the input points - # Reproject the country layer if necessary - if self.input_polygons.crs() != self.crs: - self.input_polygons = processing.run( - "native:reprojectlayer", - { - "INPUT": self.input_polygons, - "TARGET_CRS": self.crs, - "OUTPUT": "memory:", - }, - feedback=QgsProcessingFeedback(), - )["OUTPUT"] - polygon_index = QgsSpatialIndex(self.input_polygons.getFeatures()) - - # Count points within each grid cell and assign a score - reclass_vals = {} - for grid_feat in grid_layer.getFeatures(): - grid_geom = grid_feat.geometry() - # Get intersecting points - intersecting_ids = polygon_index.intersects(grid_geom.boundingBox()) - - # Initialize a set to store unique intersecting line feature IDs - unique_intersections = set() - - # Initialize variable to keep track of the maximum perimeter - max_perimeter = 0 - - for poly_id in intersecting_ids: - poly_feat = self.input_polygons.getFeature(poly_id) - poly_geom = poly_feat.geometry() - - if grid_feat.geometry().intersects(poly_geom): - unique_intersections.add(poly_id) - perimeter = poly_geom.length() - - # Update max_perimeter if this perimeter is larger - if perimeter > max_perimeter: - max_perimeter = perimeter - - # Assign reclassification value based on the maximum perimeter - if max_perimeter > 1000: # Very large blocks - reclass_val = 1 - elif 751 <= max_perimeter <= 1000: # Large blocks - reclass_val = 2 - elif 501 <= max_perimeter <= 750: # Moderate blocks - reclass_val = 3 - elif 251 <= max_perimeter <= 500: # Small blocks - reclass_val = 4 - elif 0 < max_perimeter <= 250: # Very small blocks - reclass_val = 5 - else: - reclass_val = 0 # No intersection - - reclass_vals[grid_feat.id()] = reclass_val - - # Step 5: Apply the score values to the grid - grid_layer.startEditing() - for grid_feat in grid_layer.getFeatures(): - grid_layer.changeAttributeValue( - grid_feat.id(), - provider.fieldNameIndex(field_name), - reclass_vals[grid_feat.id()], - ) - grid_layer.commitChanges() - - merged_output_vector = os.path.join(output_dir, "merged_grid_vector.shp") - - # Merge the output vector layers - merge = processing.run( - "native:mergevectorlayers", - {"LAYERS": [grid_layer], "CRS": self.crs, "OUTPUT": "TEMPORARY_OUTPUT"}, - feedback=QgsProcessingFeedback(), - )["OUTPUT"] - - xmin, xmax, ymin, ymax = ( - aligned_bbox.xMinimum(), - aligned_bbox.xMaximum(), - aligned_bbox.yMinimum(), - aligned_bbox.yMaximum(), - ) # Extent of the aligned bbox - - # Rasterize the clipped grid layer to generate the raster - rasterize_params = { - "INPUT": merge, - "FIELD": field_name, - "BURN": 0, - "USE_Z": False, - "UNITS": 1, - "WIDTH": self.pixel_size, - "HEIGHT": self.pixel_size, - "EXTENT": f"{xmin},{ymin},{xmax},{ymax}", - "NODATA": None, - "OPTIONS": "", - "DATA_TYPE": 5, # Use Int32 for scores - "OUTPUT": "TEMPORARY_OUTPUT", - } - - output_file = processing.run( - "gdal:rasterize", rasterize_params, feedback=QgsProcessingFeedback() - )["OUTPUT"] - - processing.run( - "gdal:cliprasterbymasklayer", - { - "INPUT": output_file, - "MASK": self.country_boundary, - "NODATA": -9999, - "CROP_TO_CUTLINE": True, - "OUTPUT": self.output_path, - }, - feedback=QgsProcessingFeedback(), - ) diff --git a/geest/core/polylines_per_grid_cell.py b/geest/core/polylines_per_grid_cell.py deleted file mode 100644 index 65643674..00000000 --- a/geest/core/polylines_per_grid_cell.py +++ /dev/null @@ -1,189 +0,0 @@ -import os -from qgis.PyQt.QtCore import QVariant -from qgis.core import ( - QgsGeometry, - QgsVectorLayer, - QgsField, - QgsSpatialIndex, - QgsProcessingFeedback, -) -import processing -from .utilities import GridAligner - - -class RasterPolylineGridScore: - def __init__( - self, - country_boundary, - pixel_size, - working_dir, - crs, - input_polylines, - output_path, - ): - self.country_boundary = country_boundary - self.pixel_size = pixel_size - self.working_dir = working_dir - self.crs = crs - self.input_polylines = input_polylines - self.output_path = output_path - # Initialize GridAligner with grid size - self.grid_aligner = GridAligner(grid_size=100) - - def raster_polyline_grid_score(self): - """ - Generates a raster based on the number of input points within each grid cell. - :param country_boundary: Layer defining the country boundary to clip the grid. - :param cellsize: The size of each grid cell. - :param crs: The CRS in which the grid and raster will be projected. - :param input_polylines: Layer of point features to count within each grid cell. - """ - - output_dir = os.path.dirname(self.output_path) - - # Define output directory and ensure it's created - os.makedirs(output_dir, exist_ok=True) - - # Load grid layer from the Geopackage - geopackage_path = os.path.join( - self.working_dir, "study_area", "study_area.gpkg" - ) - if not os.path.exists(geopackage_path): - raise ValueError(f"Geopackage not found at {geopackage_path}.") - - grid_layer = QgsVectorLayer( - f"{geopackage_path}|layername=study_area_grid", "merged_grid", "ogr" - ) - - area_layer = QgsVectorLayer( - f"{geopackage_path}|layername=study_area_polygons", - "study_area_polygons", - "ogr", - ) - - geometries = [feature.geometry() for feature in area_layer.getFeatures()] - - # Combine all geometries into one using unaryUnion - area_geometry = QgsGeometry.unaryUnion(geometries) - - # grid_geometry = grid_layer.getGeometry() - - aligned_bbox = self.grid_aligner.align_bbox( - area_geometry.boundingBox(), area_layer.extent() - ) - - # Extract polylines by location - grid_output = processing.run( - "native:extractbylocation", - { - "INPUT": grid_layer, - "PREDICATE": [0], - "INTERSECT": self.input_polylines, - "OUTPUT": "TEMPORARY_OUTPUT", - }, - feedback=QgsProcessingFeedback(), - )["OUTPUT"] - - grid_layer = grid_output - - # Add score field - provider = grid_layer.dataProvider() - field_name = "line_score" - if not grid_layer.fields().indexFromName(field_name) >= 0: - provider.addAttributes([QgsField(field_name, QVariant.Int)]) - grid_layer.updateFields() - - # Create spatial index for the input points - if self.input_polylines.crs() != self.crs: - self.input_polylines = processing.run( - "native:reprojectlayer", - { - "INPUT": self.input_polylines, - "TARGET_CRS": self.crs, - "OUTPUT": "memory:", - }, - feedback=QgsProcessingFeedback(), - )["OUTPUT"] - polyline_index = QgsSpatialIndex(self.input_polylines.getFeatures()) - - # Count points within each grid cell and assign a score - reclass_vals = {} - for grid_feat in grid_layer.getFeatures(): - grid_geom = grid_feat.geometry() - # Get intersecting points - intersecting_ids = polyline_index.intersects(grid_geom.boundingBox()) - - # Initialize a set to store unique intersecting line feature IDs - unique_intersections = set() - - # Check each potentially intersecting line feature - for line_id in intersecting_ids: - line_feat = self.input_polylines.getFeature(line_id) - line_geom = line_feat.geometry() - - # Perform a detailed intersection check - if grid_feat.geometry().intersects(line_geom): - unique_intersections.add(line_id) - - num_polylines = len(unique_intersections) - - # Reclassification logic: assign score based on the number of points - reclass_val = 5 if num_polylines >= 2 else 3 if num_polylines == 1 else 0 - reclass_vals[grid_feat.id()] = reclass_val - - # Apply the score values to the grid - grid_layer.startEditing() - for grid_feat in grid_layer.getFeatures(): - grid_layer.changeAttributeValue( - grid_feat.id(), - provider.fieldNameIndex(field_name), - reclass_vals[grid_feat.id()], - ) - grid_layer.commitChanges() - - # Merge the output vector layers - merge = processing.run( - "native:mergevectorlayers", - {"LAYERS": [grid_layer], "CRS": self.crs, "OUTPUT": "TEMPORARY_OUTPUT"}, - feedback=QgsProcessingFeedback(), - )["OUTPUT"] - - xmin, xmax, ymin, ymax = ( - aligned_bbox.xMinimum(), - aligned_bbox.xMaximum(), - aligned_bbox.yMinimum(), - aligned_bbox.yMaximum(), - ) # Extent of the aligned bbox - - # Rasterize the clipped grid layer to generate the raster - # output_file = os.path.join(output_dir, "rasterized_grid.tif") - rasterize_params = { - "INPUT": merge, - "FIELD": field_name, - "BURN": 0, - "USE_Z": False, - "UNITS": 1, - "WIDTH": self.pixel_size, - "HEIGHT": self.pixel_size, - "EXTENT": f"{xmin},{ymin},{xmax},{ymax}", - "NODATA": None, - "OPTIONS": "", - "DATA_TYPE": 5, # Use Int32 for scores - "OUTPUT": "TEMPORARY_OUTPUT", - } - - output_file = processing.run( - "gdal:rasterize", rasterize_params, feedback=QgsProcessingFeedback() - )["OUTPUT"] - - processing.run( - "gdal:cliprasterbymasklayer", - { - "INPUT": output_file, - "MASK": self.country_boundary, - "NODATA": -9999, - "CROP_TO_CUTLINE": True, - "OUTPUT": self.output_path, - }, - feedback=QgsProcessingFeedback(), - ) diff --git a/geest/core/rasterization.py b/geest/core/rasterization.py deleted file mode 100644 index 6ad910dd..00000000 --- a/geest/core/rasterization.py +++ /dev/null @@ -1,113 +0,0 @@ -import os -from qgis.core import QgsVectorLayer, QgsProcessingFeedback -import processing - - -class Rasterizer: - def __init__( - self, - vector_layer_path, - output_dir, - pixel_size, - utm_crs, - field=None, - dimension="default", - ): - """ - Initializes the Rasterizer class with relevant parameters. - - Args: - vector_layer_path (str): Path to the vector layer to be rasterized. - output_dir (str): Directory where the rasterized output will be saved. - pixel_size (int or float): Pixel size for the rasterized output. - utm_crs (QgsCoordinateReferenceSystem): CRS for rasterization. - field (str): The field to rasterize by (optional). If None, a burn value can be used. - dimension (str): Sub-directory within the output directory where results are saved. - """ - self.vector_layer_path = vector_layer_path - self.output_dir = os.path.normpath(output_dir) - self.dimension = dimension - self.pixel_size = pixel_size - self.utm_crs = utm_crs - self.field = field # Field for rasterization (optional) - - self.current_script_path = os.path.dirname(os.path.abspath(__file__)) - self.temp_dir = os.path.join(self.output_dir, "temp") - - self.raster_output_path = os.path.join( - self.output_dir, self.dimension, "rasterized_output.tif" - ) - - # Create necessary directories - self._setup_directories() - - # Load and preprocess the vector layer - self._load_and_preprocess_vector_layer() - - def _setup_directories(self): - """Sets up the working and temporary directories.""" - os.makedirs(os.path.join(self.output_dir, self.dimension), exist_ok=True) - os.makedirs(self.temp_dir, exist_ok=True) - - def _load_and_preprocess_vector_layer(self): - """Loads and preprocesses the vector layer, including reprojecting if necessary.""" - # Load the vector layer - self.vector_layer = QgsVectorLayer( - self.vector_layer_path, "vector_layer", "ogr" - ) - if not self.vector_layer.isValid(): - raise ValueError(f"Invalid vector layer: {self.vector_layer_path}") - - # Reproject the vector layer if necessary - if self.vector_layer.crs() != self.utm_crs: - reprojected_result = processing.run( - "native:reprojectlayer", - { - "INPUT": self.vector_layer, - "TARGET_CRS": self.utm_crs, - "OUTPUT": "memory:", - }, - feedback=QgsProcessingFeedback(), - ) - self.vector_layer = reprojected_result["OUTPUT"] - - def rasterize_vector_layer(self, nodata_value=-9999, data_type=5): - """ - Rasterizes the vector layer using the gdal:rasterize algorithm. - - Args: - nodata_value (int/float): NoData value for the output raster. - data_type (int): Data type for the raster output (default is Float32). - """ - rasterize_params = { - "INPUT": self.vector_layer, - "FIELD": self.field, # Field to use for rasterization, or None for burn value - "BURN": None if self.field else 1, # Burn value if no field is provided - "UNITS": 1, # pixel size is set in units of CRS - "WIDTH": self.pixel_size, - "HEIGHT": self.pixel_size, - "EXTENT": self.vector_layer.extent(), - "NODATA": nodata_value, - "DATA_TYPE": data_type, # Data type: Float32 (5) or others - "OUTPUT": self.raster_output_path, - } - - # Run the rasterization algorithm - rasterize_result = processing.run( - "gdal:rasterize", rasterize_params, feedback=QgsProcessingFeedback() - ) - - self.rasterized_layer = rasterize_result["OUTPUT"] - if not os.path.exists(self.rasterized_layer): - raise ValueError( - f"Rasterization failed. Output file not created. {self.rasterized_layer}" - ) - - def get_rasterized_layer_path(self): - """ - Returns the path to the rasterized output layer. - - Returns: - str: Path to the rasterized layer. - """ - return self.rasterized_layer diff --git a/geest/core/tasks/study_area.py b/geest/core/tasks/study_area.py index 57593ea4..bbddd441 100644 --- a/geest/core/tasks/study_area.py +++ b/geest/core/tasks/study_area.py @@ -37,7 +37,7 @@ class StudyAreaProcessingTask(QgsTask): It works through the (multi)part geometries in the input layer, creating bounding boxes and masks. The masks are stored as individual tif files and then a vrt file is created to combine them. The grids are in two forms - the entire bounding box and the individual parts. - The grids are aligned to 100m intervals and saved as vector features in a GeoPackage. + The grids are aligned to cell_size_m intervals and saved as vector features in a GeoPackage. Any invalid geometries are discarded, and fixed geometries are processed. Args: @@ -52,6 +52,7 @@ def __init__( name: str, layer: QgsVectorLayer, field_name: str, + cell_size_m: float, working_dir: str, mode: str = "raster", crs: Optional[QgsCoordinateReferenceSystem] = None, @@ -64,6 +65,7 @@ def __init__( :param name: The name of the task. :param layer: The vector layer containing study area features. :param field_name: The name of the field containing area names. + :param cell_size: The size of the grid cells in meters. :param working_dir: Directory path where outputs will be saved. :param mode: Processing mode, either 'vector' or 'raster'. Default is raster. :param crs: Optional CRS for the output CRS. If None, a UTM zone @@ -75,6 +77,7 @@ def __init__( self.feedback = feedback self.layer: QgsVectorLayer = layer self.field_name: str = field_name + self.cell_size_m: float = cell_size_m self.working_dir: str = working_dir self.mode: str = mode self.context: QgsProcessingContext = context @@ -127,7 +130,7 @@ def __init__( tag="Geest", level=Qgis.Info, ) - # Reproject and align the transformed layer_bbox to a 100m grid and output crs + # Reproject and align the transformed layer_bbox to a cell_size_m grid and output crs self.layer_bbox = self.grid_aligned_bbox(self.layer_bbox) def run(self) -> bool: @@ -396,8 +399,9 @@ def process_multipart_geometry( def grid_aligned_bbox(self, bbox: QgsRectangle) -> QgsRectangle: """ - Transforms and aligns the bounding box to a 100m grid in the output CRS. - The alignment ensures that the bbox aligns with the study area grid, offset by an exact multiple of 100m. + Transforms and aligns the bounding box to the grid in the output CRS. + The alignment ensures that the bbox aligns with the study area grid, offset by an exact multiple of + the grid size in m. :param bbox: The bounding box to be aligned, in the CRS of the input layer. :return: A new bounding box aligned to the grid, in the output CRS. @@ -410,33 +414,63 @@ def grid_aligned_bbox(self, bbox: QgsRectangle) -> QgsRectangle: bbox_transformed = transform.transformBoundingBox(bbox) # Align the bounding box to a grid aligned at 100m intervals, offset by the study area origin - study_area_origin_x = int(self.layer_bbox.xMinimum() // 100) * 100 - study_area_origin_y = int(self.layer_bbox.yMinimum() // 100) * 100 + study_area_origin_x = ( + int(self.layer_bbox.xMinimum() // self.cell_size_m) * self.cell_size_m + ) + study_area_origin_y = ( + int(self.layer_bbox.yMinimum() // self.cell_size_m) * self.cell_size_m + ) # Align bbox to the grid based on the study area origin x_min = ( study_area_origin_x - + int((bbox_transformed.xMinimum() - study_area_origin_x) // 100) * 100 + + int( + (bbox_transformed.xMinimum() - study_area_origin_x) // self.cell_size_m + ) + * self.cell_size_m ) y_min = ( study_area_origin_y - + int((bbox_transformed.yMinimum() - study_area_origin_y) // 100) * 100 + + int( + (bbox_transformed.yMinimum() - study_area_origin_y) // self.cell_size_m + ) + * self.cell_size_m ) x_max = ( study_area_origin_x - + (int((bbox_transformed.xMaximum() - study_area_origin_x) // 100) + 1) - * 100 + + ( + int( + (bbox_transformed.xMaximum() - study_area_origin_x) + // self.cell_size_m + ) + + 1 + ) + * self.cell_size_m ) y_max = ( study_area_origin_y - + (int((bbox_transformed.yMaximum() - study_area_origin_y) // 100) + 1) - * 100 + + ( + int( + (bbox_transformed.yMaximum() - study_area_origin_y) + // self.cell_size_m + ) + + 1 + ) + * self.cell_size_m ) - y_min -= 100 # Offset by 100m to ensure the grid covers the entire geometry - y_max += 100 # Offset by 100m to ensure the grid covers the entire geometry - x_min -= 100 # Offset by 100m to ensure the grid covers the entire geometry - x_max += 100 # Offset by 100m to ensure the grid covers the entire geometry + y_min -= ( + self.cell_size_m + ) # Offset to ensure the grid covers the entire geometry + y_max += ( + self.cell_size_m + ) # Offset to ensure the grid covers the entire geometry + x_min -= ( + self.cell_size_m + ) # Offset to ensure the grid covers the entire geometry + x_max += ( + self.cell_size_m + ) # Offset to ensure the grid covers the entire geometry # Return the aligned bbox in the output CRS return QgsRectangle(x_min, y_min, x_max, y_max) @@ -572,7 +606,7 @@ def create_and_save_grid(self, geom: QgsGeometry, bbox: QgsRectangle) -> None: ) return - step = 100 # 100m grid cells + step = self.cell_size_m # cell_size_mm grid cells feature_id += 1 feature_batch = [] @@ -696,8 +730,8 @@ def create_raster_mask( temp_layer_data_provider.addFeature(temp_feature) # Ensure resolution parameters are properly formatted as float values - x_res = 100.0 # 100m pixel size in X direction - y_res = 100.0 # 100m pixel size in Y direction + x_res = self.cell_size_m # 100m pixel size in X direction + y_res = self.cell_size_m # 100m pixel size in Y direction # Define rasterization parameters for the temporary layer params = { diff --git a/geest/core/utilities.py b/geest/core/utilities.py index 7ab39a91..ec541e98 100644 --- a/geest/core/utilities.py +++ b/geest/core/utilities.py @@ -87,95 +87,3 @@ def which(name, flags=os.X_OK): result.append(path_extensions) return result - - -def calculate_cardinality(angle): - """Compute the cardinality of an angle. - - ..versionadded: 1.0 - - ..notes: Adapted from original function with the same - name I wrote for InaSAFE. - - :param angle: Bearing angle. - :type angle: float - - :return: Cardinality text. - :rtype: str - """ - # this method could still be improved later, since the acquisition interval - # is a bit strange, i.e the input angle of 22.499° will return `N` even - # though 22.5° is the direction for `NNE` - - direction_list = ("N,NNE,NE,ENE,E,ESE,SE,SSE,S,SSW,SW,WSW,W,WNW,NW,NNW").split(",") - - bearing = float(angle) - direction_count = len(direction_list) - direction_interval = 360.0 / direction_count - index = int(floor(bearing / direction_interval)) - index %= direction_count - return direction_list[index] - - -class GridAligner: - def __init__(self, grid_size: int = 100): - """ - Initializes the GridAligner class with grid size. - - :param grid_size: The size of the grid for alignment (default is 100m). - """ - self.grid_size = grid_size # The size of the grid (default is 100m) - - def align_bbox( - self, bbox: QgsRectangle, study_area_bbox: QgsRectangle = None - ) -> QgsRectangle: - """ - Aligns the bounding box to a grid, assuming the bounding box is already in the correct CRS. - - :param bbox: The bounding box to be aligned. - :param study_area_bbox: The bounding box of the study area to define the grid origin. If None, it defaults to the bounding box itself. - :return: A new bounding box aligned to the grid. - """ - - # If no study area bbox is provided, use the bbox itself - if study_area_bbox is None: - study_area_bbox = bbox - - # Calculate the study area origin (lower-left corner) based on the provided bounding box or default to bbox - study_area_origin_x = ( - int(study_area_bbox.xMinimum() // self.grid_size) * self.grid_size - ) - study_area_origin_y = ( - int(study_area_bbox.yMinimum() // self.grid_size) * self.grid_size - ) - - # Align bbox to the grid based on the study area origin - x_min = ( - study_area_origin_x - + int((bbox.xMinimum() - study_area_origin_x) // self.grid_size) - * self.grid_size - ) - y_min = ( - study_area_origin_y - + int((bbox.yMinimum() - study_area_origin_y) // self.grid_size) - * self.grid_size - ) - x_max = ( - study_area_origin_x - + (int((bbox.xMaximum() - study_area_origin_x) // self.grid_size) + 1) - * self.grid_size - ) - y_max = ( - study_area_origin_y - + (int((bbox.yMaximum() - study_area_origin_y) // self.grid_size) + 1) - * self.grid_size - ) - - # Offset by grid size to ensure the grid covers the entire geometry - y_min -= self.grid_size - y_max += self.grid_size - x_min -= self.grid_size - x_max += self.grid_size - - # Return the aligned bbox - return QgsRectangle(x_min, y_min, x_max, y_max) diff --git a/geest/core/workflow_factory.py b/geest/core/workflow_factory.py index 2ffdac32..1a372c0f 100644 --- a/geest/core/workflow_factory.py +++ b/geest/core/workflow_factory.py @@ -33,13 +33,18 @@ class WorkflowFactory: """ def create_workflow( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Determines the workflow to return based on 'Analysis Mode' in the attributes. Passes the feedback object to the workflow for progress reporting. :param item: The JsonTreeItem object representing the task. + :param cell_size_m: The cell size in meters for the analysis. :param feedback: The QgsFeedback object for progress reporting. :param context: The QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance @@ -66,34 +71,42 @@ def create_workflow( analysis_mode = attributes.get("analysis_mode", "") if analysis_mode == "use_default_index_score": - return DefaultIndexScoreWorkflow(item, feedback, context) + return DefaultIndexScoreWorkflow(item, cell_size_m, feedback, context) elif analysis_mode == "do_not_use": - return DontUseWorkflow(item, feedback, context) + return DontUseWorkflow(item, cell_size_m, feedback, context) elif analysis_mode == "use_multi_buffer_point": - return MultiBufferDistancesWorkflow(item, feedback, context) + return MultiBufferDistancesWorkflow( + item, cell_size_m, feedback, context + ) elif analysis_mode == "use_single_buffer_point": - return SinglePointBufferWorkflow(item, feedback, context) + return SinglePointBufferWorkflow(item, cell_size_m, feedback, context) elif analysis_mode == "use_point_per_cell": - return PointPerCellWorkflow(item, feedback, context) + return PointPerCellWorkflow(item, cell_size_m, feedback, context) elif analysis_mode == "use_polyline_per_cell": - return PolylinePerCellWorkflow(item, feedback, context) + return PolylinePerCellWorkflow(item, cell_size_m, feedback, context) # TODO fix inconsistent abbreviation below for Poly elif analysis_mode == "use_poly_per_cell": - return PolygonPerCellWorkflow(item, feedback, context) + return PolygonPerCellWorkflow(item, cell_size_m, feedback, context) elif analysis_mode == "factor_aggregation": - return FactorAggregationWorkflow(item, feedback, context) + return FactorAggregationWorkflow(item, cell_size_m, feedback, context) elif analysis_mode == "dimension_aggregation": - return DimensionAggregationWorkflow(item, feedback, context) + return DimensionAggregationWorkflow( + item, cell_size_m, feedback, context + ) elif analysis_mode == "analysis_aggregation": - return AnalysisAggregationWorkflow(item, feedback, context) + return AnalysisAggregationWorkflow( + item, cell_size_m, cell_size_m, feedback, context + ) elif analysis_mode == "use_csv_to_point_layer": - return AcledImpactWorkflow(item, feedback, context) + return AcledImpactWorkflow(item, cell_size_m, feedback, context) elif analysis_mode == "use_classify_poly_into_classes": - return SafetyPolygonWorkflow(item, feedback, context) + return SafetyPolygonWorkflow(item, cell_size_m, feedback, context) elif analysis_mode == "use_nighttime_lights": - return SafetyRasterWorkflow(item, feedback, context) + return SafetyRasterWorkflow(item, cell_size_m, feedback, context) elif analysis_mode == "use_environmental_hazards": - return RasterReclassificationWorkflow(item, feedback, context) + return RasterReclassificationWorkflow( + item, cell_size_m, feedback, context + ) else: raise ValueError(f"Unknown Analysis Mode: {analysis_mode}") @@ -106,4 +119,4 @@ def create_workflow( QgsMessageLog.logMessage( traceback.format_exc(), "Geest", level=Qgis.Critical ) - return DontUseWorkflow(item, feedback, context) + return DontUseWorkflow(item, cell_size_m, feedback, context) diff --git a/geest/core/workflow_job.py b/geest/core/workflow_job.py index f858c8d7..fd2fcd90 100644 --- a/geest/core/workflow_job.py +++ b/geest/core/workflow_job.py @@ -18,7 +18,11 @@ class WorkflowJob(QgsTask): job_finished = pyqtSignal(bool) def __init__( - self, description: str, context: QgsProcessingContext, item: JsonTreeItem + self, + description: str, + context: QgsProcessingContext, + item: JsonTreeItem, + cell_size_m: float = 100.0, ): """ Initialize the workflow job. @@ -27,16 +31,21 @@ def __init__( to keep things thread safe :param item: JsonTreeItem object representing the task - this is a reference so it will update the tree directly when modified + :param cell_size_m: Cell size in meters for raster operations """ super().__init__(description) self.context = ( context # QgsProcessingContext object used to pass objects to the thread ) self._item = item # ⭐️ This is a reference - whatever you change in this item will directly update the tree + self._cell_size_m = cell_size_m # Cell size in meters for raster operations self._feedback = QgsFeedback() # Feedback object for progress and cancellation workflow_factory = WorkflowFactory() self._workflow = workflow_factory.create_workflow( - item, self._feedback, self.context + item=self._item, + cell_size_m=self._cell_size_m, + feedback=self._feedback, + context=self.context, ) # Create the workflow # TODO this raises an error... need to figure out how to connect this signal # self._workflow.progressChanged.connect(self.setProgress) diff --git a/geest/core/workflow_queue_manager.py b/geest/core/workflow_queue_manager.py index 0bd79969..bf1e28c3 100644 --- a/geest/core/workflow_queue_manager.py +++ b/geest/core/workflow_queue_manager.py @@ -45,7 +45,7 @@ def add_task(self, task: QgsTask) -> None: QgsMessageLog.logMessage(f"Task added", tag="Geest", level=Qgis.Info) return task - def add_workflow(self, item: JsonTreeItem) -> None: + def add_workflow(self, item: JsonTreeItem, cell_size_m: float) -> None: """ Add a task to the WorkflowQueue for QgsProcessingContext using the item provided. @@ -60,7 +60,12 @@ def add_workflow(self, item: JsonTreeItem) -> None: # ⭐️ Note we are passing the item reference to the WorkflowJob # any changes made to the item will be reflected in the tree directly - task = WorkflowJob(description="Geest Task", item=item, context=context) + task = WorkflowJob( + description="Geest Task", + item=item, + cell_size_m=cell_size_m, + context=context, + ) self.workflow_queue.add_job(task) QgsMessageLog.logMessage( f"Task added: {task.description()}", tag="Geest", level=Qgis.Info diff --git a/geest/core/workflows/acled_impact_workflow.py b/geest/core/workflows/acled_impact_workflow.py index e6ee70e7..03ded9f1 100644 --- a/geest/core/workflows/acled_impact_workflow.py +++ b/geest/core/workflows/acled_impact_workflow.py @@ -31,7 +31,11 @@ class AcledImpactWorkflow(WorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. @@ -40,7 +44,7 @@ def __init__( :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.workflow_name = "use_csv_to_point_layer" self.csv_file = self.attributes.get("use_csv_to_point_layer CSV File", "") diff --git a/geest/core/workflows/aggregation_workflow_base.py b/geest/core/workflows/aggregation_workflow_base.py index e8426016..d2709d13 100644 --- a/geest/core/workflows/aggregation_workflow_base.py +++ b/geest/core/workflows/aggregation_workflow_base.py @@ -1,6 +1,4 @@ import os -import shutil -import glob from qgis.core import ( QgsMessageLog, Qgis, @@ -11,8 +9,6 @@ ) from qgis.analysis import QgsRasterCalculator, QgsRasterCalculatorEntry from .workflow_base import WorkflowBase -from geest.core.convert_to_8bit import RasterConverter -from geest.utilities import resources_path from geest.core import JsonTreeItem @@ -22,16 +18,21 @@ class AggregationWorkflowBase(WorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.aggregation_attributes = None # This should be set by the child class e.g. item.getIndicatorAttributes() self.analysis_mode = self.attributes.get("analysis_mode", "") diff --git a/geest/core/workflows/analysis_aggregation_workflow.py b/geest/core/workflows/analysis_aggregation_workflow.py index 52e0fe7d..a554c03f 100644 --- a/geest/core/workflows/analysis_aggregation_workflow.py +++ b/geest/core/workflows/analysis_aggregation_workflow.py @@ -14,16 +14,21 @@ class AnalysisAggregationWorkflow(AggregationWorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.id = "geest_analysis" self.aggregation_attributes = self.item.getAnalysisAttributes() diff --git a/geest/core/workflows/default_index_score_workflow.py b/geest/core/workflows/default_index_score_workflow.py index 75fc3062..0a96d950 100644 --- a/geest/core/workflows/default_index_score_workflow.py +++ b/geest/core/workflows/default_index_score_workflow.py @@ -23,16 +23,21 @@ class DefaultIndexScoreWorkflow(WorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.workflow_name = "use_default_index_score" self.index_score = int( @@ -59,6 +64,7 @@ def _process_features_for_area( :return: Raster file path of the output. """ + _ = area_features # unused QgsMessageLog.logMessage( f"Processing area {index} score workflow", "Geest", Qgis.Info ) diff --git a/geest/core/workflows/dimension_aggregation_workflow.py b/geest/core/workflows/dimension_aggregation_workflow.py index ff4a3f62..414b3be4 100644 --- a/geest/core/workflows/dimension_aggregation_workflow.py +++ b/geest/core/workflows/dimension_aggregation_workflow.py @@ -15,16 +15,21 @@ class DimensionAggregationWorkflow(AggregationWorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.aggregation_attributes = self.item.getDimensionAttributes() self.id = self.attributes["id"].lower().replace(" ", "_") diff --git a/geest/core/workflows/dont_use_workflow.py b/geest/core/workflows/dont_use_workflow.py index df788060..f47588f4 100644 --- a/geest/core/workflows/dont_use_workflow.py +++ b/geest/core/workflows/dont_use_workflow.py @@ -16,16 +16,21 @@ class DontUseWorkflow(WorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.workflow_name = "do_not_use" self.attributes["result_file"] = "" diff --git a/geest/core/workflows/factor_aggregation_workflow.py b/geest/core/workflows/factor_aggregation_workflow.py index 4d480c08..c1310f00 100644 --- a/geest/core/workflows/factor_aggregation_workflow.py +++ b/geest/core/workflows/factor_aggregation_workflow.py @@ -13,7 +13,11 @@ class FactorAggregationWorkflow(AggregationWorkflowBase): """ def __init__( - self, item: dict, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: dict, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. @@ -22,7 +26,7 @@ def __init__( :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.aggregation_attributes = self.item.getFactorAttributes() diff --git a/geest/core/workflows/multi_buffer_distances_workflow.py b/geest/core/workflows/multi_buffer_distances_workflow.py index c6692a84..d0875e5a 100644 --- a/geest/core/workflows/multi_buffer_distances_workflow.py +++ b/geest/core/workflows/multi_buffer_distances_workflow.py @@ -41,16 +41,21 @@ class MultiBufferDistancesWorkflow(WorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param: item: Item containing workflow parameters. + :cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.workflow_name = "use_multi_buffer_point" self.distances = self.attributes.get("Multi Buffer Travel Distances", None) @@ -392,7 +397,7 @@ def _merge_layers(self, layers=None, index=None): self.workflow_directory, f"{self.layer_id}_merged_isochrones_{index}.shp" ) merge_params = { - "indicators": layers, + "LAYERS": layers, "CRS": self.target_crs, "OUTPUT": merge_output, } @@ -491,7 +496,7 @@ def _create_bands(self, layer, index): band_layers.append(smallest_layer) merge_bands_params = { - "indicators": band_layers, + "LAYERS": band_layers, "CRS": self.target_crs, "OUTPUT": output_path, } diff --git a/geest/core/workflows/point_per_cell_workflow.py b/geest/core/workflows/point_per_cell_workflow.py index d9a62119..f6375fa5 100644 --- a/geest/core/workflows/point_per_cell_workflow.py +++ b/geest/core/workflows/point_per_cell_workflow.py @@ -21,16 +21,21 @@ class PointPerCellWorkflow(WorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.workflow_name = "use_point_per_cell" diff --git a/geest/core/workflows/polygon_per_cell_workflow.py b/geest/core/workflows/polygon_per_cell_workflow.py index 0e01768a..c35e39fe 100644 --- a/geest/core/workflows/polygon_per_cell_workflow.py +++ b/geest/core/workflows/polygon_per_cell_workflow.py @@ -20,16 +20,21 @@ class PolygonPerCellWorkflow(WorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree # TODO fix inconsistent abbreviation below for Poly self.workflow_name = "use_poly_per_cell" diff --git a/geest/core/workflows/polyline_per_cell_workflow.py b/geest/core/workflows/polyline_per_cell_workflow.py index 4c07db82..9869d517 100644 --- a/geest/core/workflows/polyline_per_cell_workflow.py +++ b/geest/core/workflows/polyline_per_cell_workflow.py @@ -21,16 +21,21 @@ class PolylinePerCellWorkflow(WorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.workflow_name = "use_polyline_per_cell" diff --git a/geest/core/workflows/raster_layer_workflow.py b/geest/core/workflows/raster_layer_workflow.py index 83aecd23..2d7b2588 100644 --- a/geest/core/workflows/raster_layer_workflow.py +++ b/geest/core/workflows/raster_layer_workflow.py @@ -15,16 +15,21 @@ class RasterLayerWorkflow(WorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. :param attributes: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree def do_execute(self): diff --git a/geest/core/workflows/raster_reclassification_workflow.py b/geest/core/workflows/raster_reclassification_workflow.py index 29a38055..d932ebbd 100644 --- a/geest/core/workflows/raster_reclassification_workflow.py +++ b/geest/core/workflows/raster_reclassification_workflow.py @@ -26,16 +26,19 @@ class RasterReclassificationWorkflow(WorkflowBase): def __init__( self, item: JsonTreeItem, + cell_size_m: float, feedback: QgsFeedback, context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. + :param context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.workflow_name = "use_environmental_hazards" diff --git a/geest/core/workflows/safety_polygon_workflow.py b/geest/core/workflows/safety_polygon_workflow.py index e15bc179..23c2fbd7 100644 --- a/geest/core/workflows/safety_polygon_workflow.py +++ b/geest/core/workflows/safety_polygon_workflow.py @@ -22,16 +22,19 @@ class SafetyPolygonWorkflow(WorkflowBase): def __init__( self, item: JsonTreeItem, + cell_size_m: float, feedback: QgsFeedback, context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. + :param context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.workflow_name = "use_classify_poly_into_classes" layer_path = self.attributes.get("Classify Poly into Classes Shapefile", None) diff --git a/geest/core/workflows/safety_raster_workflow.py b/geest/core/workflows/safety_raster_workflow.py index 695d2c18..f9e3dca0 100644 --- a/geest/core/workflows/safety_raster_workflow.py +++ b/geest/core/workflows/safety_raster_workflow.py @@ -23,6 +23,7 @@ class SafetyRasterWorkflow(WorkflowBase): def __init__( self, item: JsonTreeItem, + cell_size_m: float, feedback: QgsFeedback, context: QgsProcessingContext, ): @@ -32,7 +33,7 @@ def __init__( :param feedback: QgsFeedback object for progress reporting and cancellation. """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.workflow_name = "use_nighttime_lights" layer_name = self.attributes.get("Use Nighttime Lights Raster", None) diff --git a/geest/core/workflows/single_point_buffer_workflow.py b/geest/core/workflows/single_point_buffer_workflow.py index aea1e860..2cead380 100644 --- a/geest/core/workflows/single_point_buffer_workflow.py +++ b/geest/core/workflows/single_point_buffer_workflow.py @@ -22,16 +22,21 @@ class SinglePointBufferWorkflow(WorkflowBase): """ def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: float, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: Item containing workflow parameters. + :param cell_size_m: Cell size in meters. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ super().__init__( - item, feedback, context + item, cell_size_m, feedback, context ) # ⭐️ Item is a reference - whatever you change in this item will directly update the tree self.workflow_name = "use_single_buffer_point" diff --git a/geest/core/workflows/workflow_base.py b/geest/core/workflows/workflow_base.py index f4ed7cd7..ba829070 100644 --- a/geest/core/workflows/workflow_base.py +++ b/geest/core/workflows/workflow_base.py @@ -37,15 +37,21 @@ class WorkflowBase(ABC): progressChanged = pyqtSignal(int) def __init__( - self, item: JsonTreeItem, feedback: QgsFeedback, context: QgsProcessingContext + self, + item: JsonTreeItem, + cell_size_m: 100.0, + feedback: QgsFeedback, + context: QgsProcessingContext, ): """ Initialize the workflow with attributes and feedback. - :param attributes: Item containing workflow parameters. + :param item: JsonTreeItem object representing the task. + :param cell_size_m: The cell size in meters for the analysis. :param feedback: QgsFeedback object for progress reporting and cancellation. :context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance """ self.item = item # ⭐️ This is a reference - whatever you change in this item will directly update the tree + self.cell_size_m = cell_size_m self.feedback = feedback self.context = context # QgsProcessingContext self.workflow_name = None # This is set in the concrete class @@ -57,7 +63,6 @@ def __init__( raise ValueError("Working directory not set.") # This is the lower level directory for this workflow self.workflow_directory = self._create_workflow_directory() - self.pixel_size = 100.0 # TODO get from data model self.gpkg_path: str = os.path.join( self.working_directory, "study_area", "study_area.gpkg" ) @@ -390,7 +395,7 @@ def _subset_raster_layer(self, bbox: QgsGeometry, index: int): "INPUT": self.raster_layer, "TARGET_CRS": self.target_crs, "RESAMPLING": 0, - "TARGET_RESOLUTION": self.pixel_size, + "TARGET_RESOLUTION": self.cell_size_m, "NODATA": -9999, "OUTPUT": "TEMPORARY_OUTPUT", "TARGET_EXTENT": f"{bbox.xMinimum()},{bbox.xMaximum()},{bbox.yMinimum()},{bbox.yMaximum()} [{self.target_crs.authid()}]", @@ -487,8 +492,8 @@ def _rasterize( QgsMessageLog.logMessage(f"Rasterizing {input_layer}", "Geest", Qgis.Info) # Ensure resolution parameters are properly formatted as float values - x_res = 100.0 # 100m pixel size in X direction - y_res = 100.0 # 100m pixel size in Y direction + x_res = self.cell_size_m # pixel size in X direction + y_res = self.cell_size_m # pixel size in Y direction bbox = bbox.boundingBox() # Define rasterization parameters for the temporary layer params = { @@ -546,7 +551,7 @@ def _mask_raster( output_name = f"{self.layer_id}_masked_{index}.tif" output_path = os.path.join(self.workflow_directory, output_name) QgsMessageLog.logMessage( - f"Masking raster {raster_path} with area {index} to {output_path}", + f"Masking raster {raster_path} for area {index} to {output_path}", tag="Geest", level=Qgis.Info, ) diff --git a/geest/gui/panels/setup_panel.py b/geest/gui/panels/setup_panel.py index 45b07b0b..664bf286 100644 --- a/geest/gui/panels/setup_panel.py +++ b/geest/gui/panels/setup_panel.py @@ -1,4 +1,5 @@ import os +import json import platform import shutil from PyQt5.QtWidgets import ( @@ -205,6 +206,12 @@ def create_project(self): except Exception as e: QMessageBox.critical(self, "Error", f"Failed to copy model.json: {e}") return + # open the model.json to set the analysis cell size, then close it again + with open(model_path, "r") as f: + model = json.load(f) + model["analysis_cell_size_m"] = self.cell_size_spinbox.value() + with open(model_path, "w") as f: + json.dump(model, f) # Create the processor instance and process the features debug_env = int(os.getenv("GEEST_DEBUG", 0)) @@ -217,6 +224,7 @@ def create_project(self): name="Study Area Processing", layer=layer, field_name=field_name, + cell_size_m=self.cell_size_spinbox.value(), crs=crs, working_dir=self.working_dir, context=context, diff --git a/geest/gui/panels/tree_panel.py b/geest/gui/panels/tree_panel.py index af4cbbc1..1e38eaa1 100644 --- a/geest/gui/panels/tree_panel.py +++ b/geest/gui/panels/tree_panel.py @@ -755,17 +755,18 @@ def queue_workflow_task(self, item, role): The task directly modifies the item's properties to update the tree. """ task = None + cell_size_m = 1000 if role == item.role and role == "indicator": - task = self.queue_manager.add_workflow(item) + task = self.queue_manager.add_workflow(item, cell_size_m) if role == item.role and role == "factor": item.data(3)["analysis_mode"] = "factor_aggregation" - task = self.queue_manager.add_workflow(item) + task = self.queue_manager.add_workflow(item, cell_size_m) if role == item.role and role == "dimension": item.data(3)["analysis_mode"] = "dimension_aggregation" - task = self.queue_manager.add_workflow(item) + task = self.queue_manager.add_workflow(item, cell_size_m) if role == item.role and role == "analysis": item.data(3)["analysis_mode"] = "analysis_aggregation" - task = self.queue_manager.add_workflow(item) + task = self.queue_manager.add_workflow(item, cell_size_m) if task is None: return diff --git a/geest/gui/views/treeview.py b/geest/gui/views/treeview.py index 01e9af37..1cc1ca62 100644 --- a/geest/gui/views/treeview.py +++ b/geest/gui/views/treeview.py @@ -73,6 +73,7 @@ def loadJsonData(self, json_data): # Create the 'Analysis' parent item analysis_name = json_data.get("analysis_name", "Analysis") analysis_description = json_data.get("description", "No Description") + analysis_cell_size_m = json_data.get("analysis_cell_size_m", 100.0) working_folder = json_data.get("working_folder", "Not Set") guid = json_data.get("guid", str(uuid.uuid4())) # Deserialize UUID @@ -81,6 +82,7 @@ def loadJsonData(self, json_data): "analysis_name": analysis_name, "description": analysis_description, "working_folder": working_folder, + "analysis_cell_size_m": analysis_cell_size_m, } # Create the "Analysis" item @@ -325,6 +327,7 @@ def recurse_tree(item): "analysis_name": item.data(3)["analysis_name"], "description": item.data(3)["description"], "working_folder": item.data(3)["working_folder"], + "analysis_cell_size_m": item.data(3)["analysis_cell_size_m"], "guid": item.guid, # Serialize UUID "dimensions": [recurse_tree(child) for child in item.childItems], } diff --git a/geest/ui/setup_panel_base.ui b/geest/ui/setup_panel_base.ui index 8b2b31a9..b70e33cd 100644 --- a/geest/ui/setup_panel_base.ui +++ b/geest/ui/setup_panel_base.ui @@ -190,7 +190,7 @@ - false + true 10 @@ -206,7 +206,7 @@ - false + true Analysis cell size (m) diff --git a/test/test_buffer_geom.py b/test/test_buffer_geom.py deleted file mode 100644 index 7d95047c..00000000 --- a/test/test_buffer_geom.py +++ /dev/null @@ -1,50 +0,0 @@ -import unittest -import os -from qgis.core import QgsVectorLayer, QgsCoordinateReferenceSystem -from geest.core.buffering import SinglePointBuffer - - -class TestSinglePointBuffer(unittest.TestCase): - """Test the SinglePointBuffer class.""" - - def test_create_point_buffer(self): - """ - Test the buffer creation with CRS check and reprojection. - """ - # Prepare test data - working_dir = os.path.dirname(__file__) - input_layer_path = os.path.join( - working_dir, "test_data", "points", "points.shp" - ) - output_path = os.path.join(working_dir, "output", "buffered_layer.shp") - - # Ensure output directory exists - os.makedirs(os.path.join(working_dir, "output"), exist_ok=True) - - # Load the input layer - input_layer = QgsVectorLayer(input_layer_path, "test_polygon", "ogr") - self.assertTrue(input_layer.isValid(), "The input layer is not valid.") - - # Define buffer parameters - buffer_distance = 100 # Buffer distance in CRS units - expected_crs = QgsCoordinateReferenceSystem("EPSG:32620") # UTM Zone 20N - - # Create SinglePointBuffer instance and generate the buffer - buffer_gen = SinglePointBuffer( - input_layer, buffer_distance, output_path, expected_crs - ) - buffered_layer = buffer_gen.create_buffer() - - # Check that the buffered layer file was created - self.assertTrue( - os.path.exists(output_path), "The buffered output file was not created." - ) - self.assertTrue(buffered_layer.isValid(), "The buffered layer is not valid.") - - # Clean up generated output - if os.path.exists(output_path): - os.remove(output_path) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/test_convert_to_8bit.py b/test/test_convert_to_8bit.py deleted file mode 100644 index b942e900..00000000 --- a/test/test_convert_to_8bit.py +++ /dev/null @@ -1,61 +0,0 @@ -import os -import unittest -from qgis.core import QgsApplication, QgsRasterLayer, Qgis -from geest.core.convert_to_8bit import RasterConverter - - -class TestRasterConverter(unittest.TestCase): - - @classmethod - def setUpClass(self): - - # Define paths to input and output files - self.input_raster = os.path.join( - os.path.dirname(__file__), "test_data/rasters/raster.tif" - ) - self.output_raster = os.path.join( - os.path.dirname(__file__), "output/output_raster_8bit.tif" - ) - - def test_convert_to_8bit(self): - """ - Test the convert_to_8bit method of the RasterConverter class. - """ - # Create an instance of RasterConverter - converter = RasterConverter() - - # Ensure input file exists before running the test - self.assertTrue( - os.path.exists(self.input_raster), "Input raster file does not exist" - ) - - # Check if the input image is 32-bit using QGIS API - input_layer = QgsRasterLayer(self.input_raster, "Test Raster") - self.assertTrue(input_layer.isValid(), "Raster layer is not valid") - - # Get the raster band data type - input_provider = input_layer.dataProvider() - input_band_data_type = input_provider.dataType(1) - - self.assertEqual(input_band_data_type, Qgis.Float32, "Input raster is 32-bit") - - # Run the conversion - success = converter.convert_to_8bit(self.input_raster, self.output_raster) - - # Check if the conversion was successful - self.assertTrue(success, "Raster conversion failed") - - # Check if the output image is 8-bit using QGIS API - raster_layer = QgsRasterLayer(self.output_raster, "Test Raster") - self.assertTrue(raster_layer.isValid(), "Raster layer is not valid") - - # Get the raster band data type - provider = raster_layer.dataProvider() - band_data_type = provider.dataType(1) - - # Assert if the raster data type is 8-bit - self.assertEqual(band_data_type, Qgis.Byte, "Output raster is not 8-bit") - - -if __name__ == "__main__": - unittest.main() diff --git a/test/test_create_grids.py b/test/test_create_grids.py deleted file mode 100644 index 8a413af6..00000000 --- a/test/test_create_grids.py +++ /dev/null @@ -1,52 +0,0 @@ -import unittest -import os -from qgis.core import ( - QgsVectorLayer, - QgsCoordinateReferenceSystem, - QgsApplication, - QgsRectangle, -) -from qgis.analysis import QgsNativeAlgorithms -from processing.core.Processing import Processing -from geest.core.create_grids import GridCreator - - -class TestGridCreator(unittest.TestCase): - """Test the GridCreator class.""" - - def test_create_grids(self): - """Test the create_grids method with real data to ensure the grid creation process works.""" - - # Setup parameters for the GridCreator class - self.vector_layer_path = os.path.join( - os.path.dirname(__file__), "test_data/admin/Admin0.shp" - ) - self.output_dir = os.path.join(os.path.dirname(__file__), "output") - self.merged_output_path = os.path.join(self.output_dir, "merged_grid.shp") - self.utm_crs = QgsCoordinateReferenceSystem("EPSG:32620") # UTM Zone 20N - self.h_spacing = 100 - self.v_spacing = 100 - - # Create the output directory if it doesn't exist - os.makedirs(self.output_dir, exist_ok=True) - - # Initialize the GridCreator class with the real parameters - grid_creator = GridCreator(h_spacing=self.h_spacing, v_spacing=self.v_spacing) - - # Run the grid creation process - merged_grid = grid_creator.create_grids( - self.vector_layer_path, - self.output_dir, - self.utm_crs, - self.merged_output_path, - ) - - # Check that the merged grid output file was created - self.assertTrue( - os.path.exists(self.merged_output_path), - "Merged grid output file does not exist", - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/test_crs_converter.py b/test/test_crs_converter.py deleted file mode 100644 index 30847b31..00000000 --- a/test/test_crs_converter.py +++ /dev/null @@ -1,63 +0,0 @@ -import unittest -import os -from qgis.core import QgsVectorLayer, QgsCoordinateReferenceSystem -from geest.core.crs_converter import CRSConverter - - -class TestCRSConverter(unittest.TestCase): - - def setUp(self): - """ - Setup method that runs before each test. It prepares the environment. - """ - # Define paths for test data - self.working_dir = os.path.dirname(__file__) - self.test_data_dir = os.path.join(self.working_dir, "test_data") - - # Load a test layer from the test data directory - self.layer = QgsVectorLayer( - os.path.join(self.test_data_dir, "admin/Admin0.shp"), "test_layer", "ogr" - ) - self.assertTrue(self.layer.isValid(), "Layer failed to load!") - - def test_crs_conversion(self): - """ - Test CRS conversion to a different CRS. - """ - # Create an instance of CRSConverter with the test layer - converter = CRSConverter(self.layer) - - # Convert the layer to a different CRS (EPSG:3857 "Web Mercator") - target_epsg = 3857 - reprojected_layer = converter.convert_to_crs(target_epsg) - - # Get the new CRS of the reprojected layer - new_crs = reprojected_layer.crs() - - # Check if the CRS was converted correctly - expected_crs = QgsCoordinateReferenceSystem(target_epsg) - self.assertEqual( - new_crs.authid(), expected_crs.authid(), "CRS conversion failed!" - ) - - def test_no_conversion_needed(self): - """ - Test if no conversion is performed when the layer is already in the target CRS. - """ - # Create an instance of CRSConverter with the test layer - converter = CRSConverter(self.layer) - - # Convert to the same CRS (EPSG:4326) - target_epsg = 4326 - reprojected_layer = converter.convert_to_crs(target_epsg) - - # Check that the CRS remains the same - self.assertEqual( - reprojected_layer.crs().authid(), - "EPSG:4326", - "Layer CRS should remain unchanged!", - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/test_extents.py b/test/test_extents.py deleted file mode 100644 index e28db429..00000000 --- a/test/test_extents.py +++ /dev/null @@ -1,50 +0,0 @@ -import unittest -import os -from qgis.core import ( - QgsVectorLayer, - QgsCoordinateReferenceSystem, - QgsApplication, - QgsRectangle, -) -from qgis.analysis import QgsNativeAlgorithms -from processing.core.Processing import Processing -from qgis.core import QgsProcessingFeedback -from geest.core.extents import Extents - - -class TestExtents(unittest.TestCase): - """Test the Extents class.""" - - def test_get_extent(self): - """Test the get_country_extent method to ensure extent calculation is correct.""" - - # Setup parameters for the Extents class - self.workingDir = os.path.join(os.path.dirname(__file__)) - self.vector_layer_path = os.path.join( - os.path.dirname(__file__), "test_data/admin/Admin0.shp" - ) - self.utm_crs = QgsCoordinateReferenceSystem("EPSG:32620") # UTM Zone 20N - self.pixel_size = 100 - - # Initialize the Extents class - extents_processor = Extents( - self.workingDir, self.vector_layer_path, self.pixel_size, self.utm_crs - ) - - # Get the extent of the vector layer - country_extent = extents_processor.get_country_extent() - - # Check that the extent is a valid QgsRectangle - self.assertIsInstance( - country_extent, QgsRectangle, "The extent is not a valid QgsRectangle" - ) - self.assertGreater( - country_extent.width(), 0, "Extent width is not greater than zero" - ) - self.assertGreater( - country_extent.height(), 0, "Extent height is not greater than zero" - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/test_index_score.py b/test/test_index_score.py deleted file mode 100644 index 0af4f91d..00000000 --- a/test/test_index_score.py +++ /dev/null @@ -1,96 +0,0 @@ -import unittest -import os -from qgis.core import ( - QgsVectorLayer, - QgsCoordinateReferenceSystem, - QgsProcessingFeedback, -) -import processing -from geest.core.index_score import RasterizeIndexScoreValue - - -class TestRasterizeIndexScoreValue(unittest.TestCase): - """Test the Rasterizer Index Score class.""" - - def test_generate_raster(self): - """ - Test raster generation using a small sample boundary and score value. - """ - # Prepare test data - working_dir = os.path.dirname(__file__) - boundary_path = os.path.join(working_dir, "test_data", "admin", "Admin0.shp") - output_path = os.path.join(working_dir, "output", "test_raster.tif") - - # Ensure output directory exists - os.makedirs(os.path.join(working_dir, "output"), exist_ok=True) - - # Load the country boundary layer - country_boundary = QgsVectorLayer(boundary_path, "test_boundary", "ogr") - self.assertTrue(country_boundary.isValid(), "The boundary layer is not valid.") - - crs = QgsCoordinateReferenceSystem("EPSG:32620") # UTM Zone 20N - - # Reproject the vector layer if necessary - if country_boundary.crs() != crs: - reprojected_result = processing.run( - "native:reprojectlayer", - { - "INPUT": country_boundary, - "TARGET_CRS": crs, - "OUTPUT": "memory:", - }, - feedback=QgsProcessingFeedback(), - ) - country_boundary = reprojected_result["OUTPUT"] - - # Define bbox and CRS - bbox = country_boundary.extent() - pixel_size = 100 # 100m grid - crs = QgsCoordinateReferenceSystem("EPSG:32620") # UTM Zone 20N - index_value = 80 # Index value - index_scale = 100 - - # Create RasterizeIndexScoreValue instance and generate the raster - raster_gen = RasterizeIndexScoreValue( - bbox, - country_boundary, - pixel_size, - output_path, - crs, - index_value, - index_scale, - ) - raster_layer = raster_gen.generate_raster() - - # Check that the raster file was created - self.assertTrue( - os.path.exists(output_path), "The raster output file was not created." - ) - self.assertTrue(raster_layer.isValid(), "The raster layer is not valid.") - - # Verify the checksum of the generated raster - stats = raster_layer.dataProvider().bandStatistics( - 1 - ) # Checksum for the first band - # Assert the statistics match expected values - expected_min = 4 - expected_max = 4 - expected_mean = 4 - - self.assertAlmostEqual( - stats.minimumValue, - expected_min, - msg=f"Minimum value does not match: {stats.minimumValue}", - ) - self.assertAlmostEqual( - stats.maximumValue, - expected_max, - msg=f"Maximum value does not match: {stats.maximumValue}", - ) - self.assertAlmostEqual( - stats.mean, expected_mean, msg=f"Mean value does not match: {stats.mean}" - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/test_points_per_grid_cell.py b/test/test_points_per_grid_cell.py deleted file mode 100644 index 4891303e..00000000 --- a/test/test_points_per_grid_cell.py +++ /dev/null @@ -1,87 +0,0 @@ -import unittest -import os -from qgis.core import ( - QgsVectorLayer, - QgsRasterLayer, - QgsCoordinateReferenceSystem, -) -from geest.core.points_per_grid_cell import ( - RasterPointGridScore, -) # Adjust the path to your class - - -class TestRasterPointGridScore(unittest.TestCase): - """Test the RasterPointGridScore class.""" - - def test_raster_point_grid_score(self): - """ - Test raster generation using the RasterPointGridScore class. - """ - self.working_dir = os.path.dirname(__file__) - self.test_data_dir = os.path.join(self.working_dir, "test_data") - - # Load the input data (points and country boundary layers) - self.point_layer = QgsVectorLayer( - os.path.join(self.test_data_dir, "points/points.shp"), "test_points", "ogr" - ) - self.country_boundary = os.path.join(self.test_data_dir, "admin/Admin0.shp") - - self.assertTrue(self.point_layer.isValid(), "The point layer is not valid.") - - # Define output path for the generated raster - self.output_path = os.path.join( - self.working_dir, "output", "test_points_per_grid_cell.tif" - ) - os.makedirs(os.path.join(self.working_dir, "output"), exist_ok=True) - - # Define CRS (for example UTM Zone 20N) - self.crs = QgsCoordinateReferenceSystem("EPSG:32620") - self.pixel_size = 100 # 100m grid - - # Create an instance of the RasterPointGridScore class - rasterizer = RasterPointGridScore( - country_boundary=self.country_boundary, - pixel_size=self.pixel_size, - output_path=self.output_path, - crs=self.crs, - input_points=self.point_layer, - ) - - # Run the raster_point_grid_score method - rasterizer.raster_point_grid_score() - - # Load the generated raster layer to verify its validity - # Verify that the raster file was created - self.assertTrue( - os.path.exists(self.output_path), "The raster output file was not created." - ) - raster_layer = QgsRasterLayer(self.output_path, "test_raster", "gdal") - self.assertTrue( - raster_layer.isValid(), "The generated raster layer is not valid." - ) - - # Verify raster statistics (e.g., minimum, maximum, mean) - stats = raster_layer.dataProvider().bandStatistics( - 1 - ) # Get statistics for the first band - expected_min = ( - 0 # Update this with the actual expected value based on your data - ) - expected_max = ( - 3 # Update this with the actual expected value based on your data - ) - - self.assertAlmostEqual( - stats.minimumValue, - expected_min, - msg=f"Minimum value does not match: {stats.minimumValue}", - ) - self.assertAlmostEqual( - stats.maximumValue, - expected_max, - msg=f"Maximum value does not match: {stats.maximumValue}", - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/test_polygons_per_grid_cell.py b/test/test_polygons_per_grid_cell.py deleted file mode 100644 index c85e4553..00000000 --- a/test/test_polygons_per_grid_cell.py +++ /dev/null @@ -1,91 +0,0 @@ -import unittest -import os -from qgis.core import ( - QgsVectorLayer, - QgsRasterLayer, - QgsCoordinateReferenceSystem, -) -from geest.core.polygons_per_grid_cell import ( - RasterPolygonGridScore, -) # Adjust the path to your class - - -class TestRasterPolygonGridScore(unittest.TestCase): - """Test the RasterPolygonGridScore class.""" - - def test_raster_polygon_grid_score(self): - """ - Test raster generation using the RasterPolygonGridScore class. - """ - self.working_dir = os.path.dirname(__file__) - self.test_data_dir = os.path.join(self.working_dir, "test_data") - os.chdir(self.working_dir) - - # Load the input data (polygons and country boundary layers) - self.polygon_layer = QgsVectorLayer( - os.path.join(self.test_data_dir, "polygons", "blocks.shp"), - "test_polygons", - "ogr", - ) - self.country_boundary = os.path.join(self.test_data_dir, "admin", "Admin0.shp") - - self.assertTrue(self.polygon_layer.isValid(), "The polygon layer is not valid.") - - # Define output path for the generated raster - self.output_path = os.path.join( - self.working_dir, "output", "test_polygons_per_grid_cell.tif" - ) - os.makedirs(os.path.join(self.working_dir, "output"), exist_ok=True) - - # Define CRS (for example UTM Zone 20N) - self.crs = QgsCoordinateReferenceSystem("EPSG:32620") - self.pixel_size = 100 # 100m grid - - # Create an instance of the RasterPolygonGridScore class - rasterizer = RasterPolygonGridScore( - country_boundary=self.country_boundary, - pixel_size=self.pixel_size, - working_dir=self.test_data_dir, - crs=self.crs, - input_polygons=self.polygon_layer, - output_path=self.output_path, - ) - - # Run the raster_polygon_grid_score method - rasterizer.raster_polygon_grid_score() - - # Load the generated raster layer to verify its validity - # Verify that the raster file was created - self.assertTrue( - os.path.exists(self.output_path), "The raster output file was not created." - ) - raster_layer = QgsRasterLayer(self.output_path, "test_raster", "gdal") - self.assertTrue( - raster_layer.isValid(), "The generated raster layer is not valid." - ) - - # Verify raster statistics (e.g., minimum, maximum, mean) - stats = raster_layer.dataProvider().bandStatistics( - 1 - ) # Get statistics for the first band - expected_min = ( - 0 # Update this with the actual expected value based on your data - ) - expected_max = ( - 5 # Update this with the actual expected value based on your data - ) - - self.assertAlmostEqual( - stats.minimumValue, - expected_min, - msg=f"Minimum value does not match: {stats.minimumValue}", - ) - self.assertAlmostEqual( - stats.maximumValue, - expected_max, - msg=f"Maximum value does not match: {stats.maximumValue}", - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/test_polylines_per_grid_cell.py b/test/test_polylines_per_grid_cell.py deleted file mode 100644 index c3ce81d1..00000000 --- a/test/test_polylines_per_grid_cell.py +++ /dev/null @@ -1,95 +0,0 @@ -import unittest -import os -from qgis.core import ( - QgsVectorLayer, - QgsRasterLayer, - QgsCoordinateReferenceSystem, -) -from geest.core.polylines_per_grid_cell import ( - RasterPolylineGridScore, -) - - -class TestRasterPolylineGridScore(unittest.TestCase): - """Test the RasterPolylineGridScore class.""" - - def test_raster_polyline_grid_score(self): - """ - Test raster generation using the RasterPolylineGridScore class. - """ - self.working_dir = os.path.dirname(__file__) - self.test_data_dir = os.path.join(self.working_dir, "test_data") - os.chdir(self.working_dir) - - # Load the input data (polylines and country boundary layers) - self.polyline_layer = QgsVectorLayer( - os.path.join(self.test_data_dir, "polylines", "polylines.shp"), - "test_polylines", - "ogr", - ) - self.country_boundary = os.path.join(self.test_data_dir, "admin", "Admin0.shp") - - self.assertTrue( - self.polyline_layer.isValid(), "The polyline layer is not valid." - ) - - # Define output path for the generated raster - self.output_path = os.path.join( - self.working_dir, "output", "rasterized_grid.tif" - ) - os.makedirs(os.path.join(self.working_dir, "output"), exist_ok=True) - - # Define CRS (for example UTM Zone 20N) - self.crs = QgsCoordinateReferenceSystem("EPSG:32620") - self.pixel_size = 100 # 100m grid - - # Create an instance of the RasterPolylineGridScore class - rasterizer = RasterPolylineGridScore( - country_boundary=self.country_boundary, - pixel_size=self.pixel_size, - working_dir=self.test_data_dir, - crs=self.crs, - input_polylines=self.polyline_layer, - output_path=self.output_path, - ) - - # Run the raster_polyline_grid_score method - rasterizer.raster_polyline_grid_score() - - # Load the generated raster layer to verify its validity - self.assertTrue( - os.path.exists(self.output_path), "The raster output file was not created." - ) - # self.clipped_output_path = os.path.join( - # self.working_dir, "output", "clipped_rasterized_grid.tif" - # ) - raster_layer = QgsRasterLayer(self.output_path, "test_raster", "gdal") - self.assertTrue( - raster_layer.isValid(), "The generated raster layer is not valid." - ) - - # Verify raster statistics (e.g., minimum, maximum, mean) - stats = raster_layer.dataProvider().bandStatistics( - 1 - ) # Get statistics for the first band - expected_min = ( - 0 # Update this with the actual expected value based on your data - ) - expected_max = ( - 5 # Update this with the actual expected value based on your data - ) - - self.assertAlmostEqual( - stats.minimumValue, - expected_min, - msg=f"Minimum value does not match: {stats.minimumValue}", - ) - self.assertAlmostEqual( - stats.maximumValue, - expected_max, - msg=f"Maximum value does not match: {stats.maximumValue}", - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/test_rasterizer.py b/test/test_rasterizer.py deleted file mode 100644 index 88682070..00000000 --- a/test/test_rasterizer.py +++ /dev/null @@ -1,58 +0,0 @@ -import unittest -import os -from qgis.core import QgsVectorLayer, QgsCoordinateReferenceSystem, QgsApplication -from qgis.analysis import QgsNativeAlgorithms -from processing.core.Processing import Processing -from qgis.core import QgsProcessingFeedback -from geest.core.rasterization import Rasterizer - - -class TestRasterizer(unittest.TestCase): - """Test the Rasterizer class.""" - - def setUp(self): - # Setup real parameters for the Rasterizer class - self.vector_layer_path = os.path.join( - os.path.dirname(__file__), "test_data/admin/Admin0.shp" - ) - self.output_dir = os.path.join(os.path.dirname(__file__), "output") - self.pixel_size = 100 - self.utm_crs = QgsCoordinateReferenceSystem("EPSG:32620") # UTM Zone 20N - self.field = "score" - self.dimension = "Contextual" - - # Create the output directory if it doesn't exist - os.makedirs(self.output_dir, exist_ok=True) - - def test_rasterize_vector_layer(self): - """ - Test the rasterize_vector_layer method with real data to ensure the rasterization process works. - """ - # Initialize the Rasterizer class with real parameters - rasterizer = Rasterizer( - vector_layer_path=self.vector_layer_path, - output_dir=self.output_dir, - pixel_size=self.pixel_size, - utm_crs=self.utm_crs, - field=self.field, - dimension=self.dimension, - ) - - # Load the vector layer and check if it is valid - rasterizer._load_and_preprocess_vector_layer() # Assuming this method exists to load the layer - self.assertTrue( - rasterizer.vector_layer.isValid(), "The vector layer is not valid" - ) - - # Run the real rasterization process - rasterizer.rasterize_vector_layer() - - # Check that the rasterized output file was created - rasterized_output = rasterizer.get_rasterized_layer_path() - self.assertTrue( - os.path.exists(rasterized_output), "Rasterized output file does not exist" - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/test/test_safety_per_cell_processor.py b/test/test_safety_per_cell_processor.py index 4fca15bb..dc45f29a 100644 --- a/test/test_safety_per_cell_processor.py +++ b/test/test_safety_per_cell_processor.py @@ -54,6 +54,7 @@ def setUp(self): output_prefix="test", safety_layer=self.safety_layer, safety_field="safety", + cell_size_m=100.0, workflow_directory=self.workflow_directory, gpkg_path=self.gpkg_path, context=self.context,