schema 300 leftovers

CHANGES.rst

@@ -3,11 +3,21 @@ Changelog of threedi-modelchecker
 
 
 
-2.14.2 (unreleased)
+2.15.0 (unreleased)
 -------------------
 
-- Add test for check descriptions.
 - Change minimum python version to 3.9 in pyproject.toml, update test matrix.
+- Check if tables related to use_* settings in model_settings and simulation_template settings are populated
+- Warn if tables related to use_* settings in model_settings and simulation_template settings are populated while use_* settings is false
+- Add test for check descriptions.
+- Collect all foreign key checks and give them a uniform error or warning (0001)
+- Add unique check for boundary_condition_1d.connection_node_id
+- Add checks for dry_weather_flow_distribution.distribution format, length and sum
+- Add check if geometries for orifice, weir and pipe match their connection nodes
+- Add check if geometries for control_measure_map, dry_weather_flow_map, surface_map and pump_map match the object they connect
+- Add check if windshielding geometry matches with that of the linked channel
+- Add check if the geometry of boundary_condition_1d, control_measure_location, lateral_1d, and pump matches with that of the linked connection node
+- Add check if the geometry of memory_control or table_control matches to that of the linked object
 
 
 2.14.1 (2024-11-25)

pyproject.toml

@@ -19,7 +19,7 @@ dependencies = [
     "Click",
     "GeoAlchemy2>=0.9,!=0.11.*",
     "SQLAlchemy>=1.4",
-    "threedi-schema==0.228.*"
+    "threedi-schema==0.229.*"
 ]
 
 [project.optional-dependencies]

threedi_modelchecker/__init__.py

@@ -1,5 +1,5 @@
 from .model_checks import *  # NOQA
 
 # fmt: off
-__version__ = '2.14.2.dev0'
+__version__ = '2.15.0.dev0'
 # fmt: on

threedi_modelchecker/checks/factories.py

@@ -1,3 +1,6 @@
+from dataclasses import dataclass
+from typing import Optional
+
 from threedi_schema import custom_types
 
 from .base import (
@@ -11,33 +14,56 @@
 )
 
 
+@dataclass
+class ForeignKeyCheckSetting:
+    col: str
+    ref: str
+    filter: Optional[bool] = None
+
+
 def get_level(table, column, level_map):
     level = level_map.get(f"*.{column.name}")
     level = level_map.get(f"{table.name}.{column.name}", level)
     return level or "ERROR"
 
 
-def generate_foreign_key_checks(table, custom_level_map=None, **kwargs):
+def generate_foreign_key_checks(table, fk_settings, custom_level_map=None, **kwargs):
     custom_level_map = custom_level_map or {}
     foreign_key_checks = []
-    for fk_column in table.foreign_keys:
-        level = get_level(table, fk_column.parent, custom_level_map)
+    for fk_setting in fk_settings:
+        if fk_setting.col.table != table:
+            continue
+        level = get_level(table, fk_setting.col, custom_level_map)
+        # Prevent clash when kwargs contains 'filter'
+        filter_val = (
+            kwargs.get("filter") if fk_setting.filter is None else fk_setting.filter
+        )
+        kwargs.pop("filter", None)
         foreign_key_checks.append(
             ForeignKeyCheck(
-                reference_column=fk_column.column,
-                column=fk_column.parent,
+                reference_column=fk_setting.ref,
+                column=fk_setting.col,
                 level=level,
+                filters=filter_val,
                 **kwargs,
             )
         )
     return foreign_key_checks
 
 
-def generate_unique_checks(table, custom_level_map=None, **kwargs):
+def generate_unique_checks(
+    table, custom_level_map=None, extra_unique_columns=None, **kwargs
+):
     custom_level_map = custom_level_map or {}
     unique_checks = []
+    if extra_unique_columns is None:
+        extra_unique_columns = []
     for column in table.columns:
-        if column.unique or column.primary_key:
+        if (
+            column.unique
+            or column.primary_key
+            or any(col.compare(column) for col in extra_unique_columns)
+        ):
             level = get_level(table, column, custom_level_map)
             unique_checks.append(UniqueCheck(column, level=level, **kwargs))
     return unique_checks

threedi_modelchecker/checks/location.py

@@ -0,0 +1,166 @@
+from typing import List, NamedTuple
+
+from sqlalchemy import func
+from sqlalchemy.orm import aliased, Session
+from threedi_schema.domain import models
+
+from threedi_modelchecker.checks.base import BaseCheck
+from threedi_modelchecker.checks.geo_query import distance
+
+
+class PointLocationCheck(BaseCheck):
+    """Check if cross section locations are within {max_distance} of their channel."""
+
+    def __init__(
+        self,
+        ref_column,
+        ref_table,
+        max_distance,
+        *args,
+        **kwargs,
+    ):
+        self.max_distance = max_distance
+        self.ref_column = ref_column
+        self.ref_table = ref_table
+        super().__init__(*args, **kwargs)
+
+    def get_invalid(self, session):
+        # get all channels with more than 1 cross section location
+        return (
+            self.to_check(session)
+            .join(
+                self.ref_table,
+                self.ref_table.id == self.ref_column,
+            )
+            .filter(distance(self.column, self.ref_table.geom) > self.max_distance)
+            .all()
+        )
+
+    def description(self):
+        return (
+            f"{self.column_name} does not match the position of the object that "
+            f"{self.table.name}.{self.ref_column} refers to"
+        )
+
+
+class LinestringLocationCheck(BaseCheck):
+    """Check that linestring geometry starts / ends are close to their connection nodes
+
+    This allows for reversing the geometries. threedi-gridbuilder will reverse the geometries if
+    that lowers the distance to the connection nodes.
+    """
+
+    def __init__(
+        self,
+        ref_column_start,
+        ref_column_end,
+        ref_table_start,
+        ref_table_end,
+        max_distance,
+        *args,
+        **kwargs,
+    ):
+        self.max_distance = max_distance
+        self.ref_column_start = ref_column_start
+        self.ref_column_end = ref_column_end
+        self.ref_table_start = ref_table_start
+        self.ref_table_end = ref_table_end
+        super().__init__(*args, **kwargs)
+
+    def get_invalid(self, session: Session) -> List[NamedTuple]:
+        start_node = aliased(self.ref_table_start)
+        end_node = aliased(self.ref_table_end)
+
+        tol = self.max_distance
+        start_point = func.ST_PointN(self.column, 1)
+        end_point = func.ST_PointN(self.column, func.ST_NPoints(self.column))
+
+        start_ok = distance(start_point, start_node.geom) <= tol
+        end_ok = distance(end_point, end_node.geom) <= tol
+        start_ok_if_reversed = distance(end_point, start_node.geom) <= tol
+        end_ok_if_reversed = distance(start_point, end_node.geom) <= tol
+        return (
+            self.to_check(session)
+            .join(start_node, start_node.id == self.ref_column_start)
+            .join(end_node, end_node.id == self.ref_column_end)
+            .filter(
+                ~(start_ok & end_ok),
+                ~(start_ok_if_reversed & end_ok_if_reversed),
+            )
+            .all()
+        )
+
+    def description(self) -> str:
+        ref_start_name = f"{self.table.name}.{self.ref_column_start.name}"
+        ref_end_name = f"{self.table.name}.{self.ref_column_end.name}"
+        return f"{self.column_name} does not start or end at its connection nodes: {ref_start_name} and {ref_end_name} (tolerance = {self.max_distance} m)"
+
+
+class ConnectionNodeLinestringLocationCheck(LinestringLocationCheck):
+    def __init__(self, column, *args, **kwargs):
+        table = column.table
+        super().__init__(
+            ref_column_start=table.c.connection_node_id_start,
+            ref_column_end=table.c.connection_node_id_end,
+            ref_table_start=models.ConnectionNode,
+            ref_table_end=models.ConnectionNode,
+            column=column,
+            *args,
+            **kwargs,
+        )
+
+    def description(self) -> str:
+        return f"{self.column_name} does not start or end at its connection node (tolerance = {self.max_distance} m)"
+
+
+class ControlMeasureMapLinestringMapLocationCheck(LinestringLocationCheck):
+    def __init__(self, control_table, filters, *args, **kwargs):
+        super().__init__(
+            ref_column_start=models.ControlMeasureMap.measure_location_id,
+            ref_column_end=models.ControlMeasureMap.control_id,
+            ref_table_start=models.ControlMeasureLocation,
+            ref_table_end=control_table,
+            column=models.ControlMeasureMap.geom,
+            filters=filters,
+            *args,
+            **kwargs,
+        )
+
+
+class DWFMapLinestringLocationCheck(LinestringLocationCheck):
+    def __init__(self, *args, **kwargs):
+        super().__init__(
+            ref_column_start=models.DryWeatherFlowMap.connection_node_id,
+            ref_column_end=models.DryWeatherFlowMap.dry_weather_flow_id,
+            ref_table_start=models.ConnectionNode,
+            ref_table_end=models.DryWeatherFlow,
+            column=models.DryWeatherFlowMap.geom,
+            *args,
+            **kwargs,
+        )
+
+
+class PumpMapLinestringLocationCheck(LinestringLocationCheck):
+    def __init__(self, *args, **kwargs):
+        super().__init__(
+            ref_column_start=models.PumpMap.pump_id,
+            ref_column_end=models.PumpMap.connection_node_id_end,
+            ref_table_start=models.Pump,
+            ref_table_end=models.ConnectionNode,
+            column=models.PumpMap.geom,
+            *args,
+            **kwargs,
+        )
+
+
+class SurfaceMapLinestringLocationCheck(LinestringLocationCheck):
+    def __init__(self, *args, **kwargs):
+        super().__init__(
+            ref_column_start=models.SurfaceMap.surface_id,
+            ref_column_end=models.SurfaceMap.connection_node_id,
+            ref_table_start=models.Surface,
+            ref_table_end=models.ConnectionNode,
+            column=models.SurfaceMap.geom,
+            *args,
+            **kwargs,
+        )