Resolve "Introduce a new filter for when rules should be run based on the source ID of a building."

Closes #21 (closed)

\approve @laurens @ds @chengzhi

rulelib/abstractrule.py

@@ -18,7 +18,6 @@
 
 
 import abc
-from typing import Union
 
 import shapely
 
@@ -29,24 +28,34 @@ class AbstractRule(abc.ABC):
     structure that is or could be necessary for all rules.
 
     Args:
+        rule_source_ids (list):
+            A list of all valid source IDs for the rule. This is provided by the rule's XML file
+            if relevant.
         geographic_filter_boundary (str):
             Boundary polygon in WKT format defining the geographic boundary of the rule.
     """
 
-    def __init__(self, geographic_filter_boundary: str = None):
+    def __init__(
+        self,
+        rule_source_ids: list | None = None,
+        geographic_filter_boundary: str | None = None,
+    ):
+
         self.geographic_filter_boundary_geometry = None
         if geographic_filter_boundary is not None:
-            self.geographic_filter_boundary_geometry = self.read_geometry_from_wkt(
+            self.geographic_filter_boundary_geometry = shapely.from_wkt(
                 geographic_filter_boundary
             )
 
+        self.rule_source_ids = rule_source_ids
+
     @abc.abstractmethod
     def __call__(self, *args, **kwargs):
         """
         The `__call__` method implements the core process of the rule.
         """
 
-    def parse_data(self, data: Union[str, bytes], source_file: str):
+    def parse_data(self, data: str | bytes, source_file: str):
         """
         The `parse_data` method can be used to parse files that are attached to the
         rule. As file types can widely vary, this method needs to be implemented for each
@@ -67,41 +76,43 @@ class AbstractRule(abc.ABC):
             + "please implement this class first."
         )
 
-    @staticmethod
-    def read_geometry_from_wkt(geographic_filter_boundary):
-        """
-        Converts a WKT boundary into a Shapely polygon.
-
-        Args:
-            geographic_filter_boundary (str):
-                Boundary polygon in WKT format defining the geographic boundary of the rule.
-
-        Returns:
-            A Shapely polygon based on the geometry.
-        """
-
-        return shapely.from_wkt(geographic_filter_boundary)
-
-    def filter(self, longitude=None, latitude=None, *args, **kwargs):
+    def filter(
+        self,
+        longitude: float | None = None,
+        latitude: float | None = None,
+        source_id: int | None = None,
+        *args,
+        **kwargs
+    ):
         """
         Applies a spatial filter to ensure that only buildings within the provided geographic
-        polygon are processed.
+        polygon are processed and a source filter to ensure source-specific rules only run
+        for buildings with a matching source ID.
 
         Args:
             longitude (float):
                 Longitude of the building being processed.
             latitude (float):
                 Latitude of the building being processed.
+            source_id (int):
+                Integer representing the building-source ID.
 
         Returns:
-            True if the point is in the boundary and False if it isn't, which skips the rule.
+            bool:
+               True if all tests pass and False if at least one does not pass.
         """
 
-        if self.geographic_filter_boundary_geometry is None:
-            return True
-        elif self.geographic_filter_boundary_geometry.contains(
-            shapely.geometry.Point(longitude, latitude)
+        # Geographic filter, passes if the building's coordinates lie within the boundary
+        # geometry.
+        if (
+            self.geographic_filter_boundary_geometry is not None
+            and self.geographic_filter_boundary_geometry.disjoint(
+                shapely.geometry.Point(longitude, latitude)
+            )
         ):
-            return True
-        else:
             return False
+
+        # Source ID filter, passes if the rule's source ID and the building-source ID match.
+        if self.rule_source_ids is not None and source_id not in self.rule_source_ids:
+            return False
+        return True