add tunnel and bridge info to os open road links

scripts/osopen_bridge_width.py → scripts/calculate_bridge.py

@@ -18,9 +18,7 @@
 
 # %%
 # inputs
-os_links = gpd.read_parquet(
-    base_path / "networks" / "road" / "GB_road_link_file.geoparquet"
-)
+os_links = gpd.read_parquet(base_path / "networks" / "road" / "GB_road_link_file.pq")
 masteros_links = gpd.read_parquet(
     base_path / "networks" / "road" / "GB_RoadLinks.pq"
 )  # averageWidth
@@ -31,21 +29,15 @@
 
 # %%
 # os link: a list of masteros links with bridge
-masteros_2_os = lut_bridge.set_index("Roadlink_Id")[
-    "OSOpenRoads_RoadLinkIdentifier"
-].to_dict()
-masteros_links["OSOpenRoads_RoadLinkIdentifier"] = masteros_links.gml_id.map(
-    masteros_2_os
+masteros_links["Roadlink_Id"] = masteros_links["gml_id"]
+lut_bridge = lut_bridge.merge(
+    masteros_links[["Roadlink_Id", "averageWidth"]], on="Roadlink_Id", how="left"
 )
-masteros_bridges = masteros_links[
-    masteros_links.OSOpenRoads_RoadLinkIdentifier.notnull()
-][["gml_id", "OSOpenRoads_RoadLinkIdentifier", "averageWidth"]]
 
-masteros_bridges_gp = masteros_bridges.groupby(
-    by=["OSOpenRoads_RoadLinkIdentifier"], as_index=False
+masteros_bridges_gp = lut_bridge.groupby(
+    by="OSOpenRoads_RoadLinkIdentifier", as_index=False
 ).agg(
     {
-        # "gml_id": list,
         "averageWidth": list,
     }
 )
@@ -55,34 +47,33 @@
 os_links_bridge = os_links.merge(
     masteros_bridges_gp, how="right", on="OSOpenRoads_RoadLinkIdentifier"
 )
-
 # sum or average?
 # collapsed dual carriageways (sum of two random bridge widths)
 # others (the average of all bridge widths)
 os_links_bridge["est_averageWidth"] = os_links_bridge.apply(
     lambda row: (
-        np.sum(sorted(row["averageWidth_y"], reverse=True)[:2])
+        np.sum(sorted(row["averageWidth"], reverse=True)[:2])
         if row["form_of_way"] == "Collapsed Dual Carriageway"
-        else np.mean(row["averageWidth_y"])
+        else np.mean(row["averageWidth"])
     ),
     axis=1,
 )
 
 os_links = os_links.merge(
     os_links_bridge[
-        ["OSOpenRoads_RoadLinkIdentifier", "averageWidth_y", "est_averageWidth"]
+        ["OSOpenRoads_RoadLinkIdentifier", "averageWidth", "est_averageWidth"]
     ],
     how="left",
     on="OSOpenRoads_RoadLinkIdentifier",
 )
-os_links.drop(
-    columns=["averageWidth", "minimumWidth", "max_z", "min_z", "mean_z"], inplace=True
-)
+
+os_links.drop(columns="averageWidth", inplace=True)
 os_links.rename(
     columns={
-        "averageWidth_y": "list_of_bridge_width",
         "est_averageWidth": "aveBridgeWidth",
     },
     inplace=True,
 )
-os_links.to_parquet(base_path / "networks" / "road" / "GB_road_link_file_bridge.pq")
+
+os_links.aveBridgeWidth = os_links.aveBridgeWidth.fillna(0)
+# os_links.to_parquet(base_path / "networks" / "road" / "GB_road_link_file.pq")

scripts/calculate_tunnel.py

@@ -0,0 +1,51 @@
+# %%
+from pathlib import Path
+import pandas as pd
+import geopandas as gpd  # type: ignore
+
+from nird.utils import load_config
+
+import warnings
+
+warnings.simplefilter("ignore")
+
+base_path = Path(load_config()["paths"]["base_path"])
+
+# %%
+lut = pd.read_csv(base_path / "tables" / "OSOpenRoadLookUpTable_major_roads.csv")
+openos_links = gpd.read_parquet(
+    base_path / "networks" / "road" / "GB_road_link_file.geoparquet"
+)
+masteros_links = gpd.read_parquet(base_path / "networks" / "road" / "GB_RoadLinks.pq")
+
+# %%
+open_to_master = lut.groupby(by="OSOpenRoads_RoadLinkIdentifier", as_index=False).agg(
+    {"Roadlink_Id": list}
+)
+master_to_tunnel = masteros_links[["gml_id", "roadStructure"]]
+master_to_tunnel["hasTunnel"] = 0
+master_to_tunnel.loc[master_to_tunnel.roadStructure.notnull(), "hasTunnel"] = 1
+master_has_tunnel = master_to_tunnel.set_index("gml_id")["hasTunnel"].to_dict()
+open_to_master["hasTunnel"] = open_to_master.apply(
+    lambda row: (
+        1
+        if any(master_has_tunnel.get(value, 0) == 1 for value in row["Roadlink_Id"])
+        else 0
+    ),
+    axis=1,
+)
+
+# %%
+openos_links = openos_links.merge(
+    open_to_master[["OSOpenRoads_RoadLinkIdentifier", "hasTunnel"]],
+    how="left",
+    on="OSOpenRoads_RoadLinkIdentifier",
+)
+
+openos_links.drop(
+    columns=["averageWidth", "minimumWidth", "max_z", "min_z", "mean_z"], inplace=True
+)
+
+openos_links["hasTunnel"] = openos_links["hasTunnel"].fillna(0).astype(int)
+# openos_links.to_parquet(base_path / "networks" / "road" / "GB_road_link_file.pq")
+# 06CC7473-D5CE-4BAF-B723-4EFF5E3C7ED4: os open: no, masteros: yes