fix(roadworks): only push roadworks with geom_type LineString to postgis

pipeline/assets/traffic_incidents.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import warnings
 
@@ -24,6 +25,7 @@
 # check when migrating to newer versions
 warnings.filterwarnings('ignore', category=ExperimentalWarning)
 
+logger = logging.getLogger(__name__)
 
 @asset(
     compute_kind='DATEX2',
@@ -71,6 +73,8 @@ def roadworks_geojson() -> dict:
     """
     Transforms roadworks datasets into GeoJSON (mapping waze cifs to a commonly diggestable format)
     and publishes them.
+    Note: these may include roadworks with geometry type point. A point geometry type
+    is not recommended as downstream standards like e.g. CIFS can't handle them.
     """
     source = os.path.join(WEB_ROOT, *ROADWORKS_ASSET_KEY_PREFIX, ROADWORKS_DATEXII_FIILENAME)
     return DatexII2CifsTransformer('MobiData BW').transform(source, 'geojson')
@@ -82,8 +86,16 @@ def roadworks_geojson() -> dict:
     io_manager_key='pg_gpd_io_manager',
     auto_materialize_policy=AutoMaterializePolicy.eager(),
 )
-def roadworks(roadworks_geojson) -> pd.DataFrame:
+def roadworks(roadworks_geojson: dict[str, any]) -> pd.DataFrame:
     """
     Imports the roadworks into PostGIS, from where they can be accessed e.g. via WMS/WFS.
+    Note: only roadworks with geom_type are imported!
     """
-    return gpd.GeoDataFrame.from_features(roadworks_geojson['features'], crs='epsg:4326')
+    roadworks_gdf = gpd.GeoDataFrame.from_features(roadworks_geojson['features'], crs='epsg:4326')
+    roadworks_not_linestrings = roadworks_gdf[roadworks_gdf.geom_type != 'LineString']
+    if len(roadworks_not_linestrings) > 0:
+        logger.warn(f'''Ignored {len(roadworks_not_linestrings)} which had
+            geom_type!=LineString, e.g. with id="{roadworks_not_linestrings["id"].iloc[0]}''')
+        roadworks_only_linestrings_gdf = roadworks_gdf[roadworks_gdf.geom_type == 'LineString']
+        return roadworks_only_linestrings_gdf.set_index('id')
+    return roadworks_gdf.set_index('id')