Merge pull request #6 from nihar1024/main

Implement network scenario support, refactor street network fetch code

src/core/config.py

@@ -13,13 +13,24 @@ class SyncPostgresDsn(PostgresDsn):
 
 
 class Settings(BaseSettings):
+    DEBUG_MODE: bool = True
+
+    CUSTOMER_SCHEMA: str = "customer"
+    USER_DATA_SCHEMA: str = "user_data"
+
     API_V2_STR: str = "/api/v2"
     PROJECT_NAME: Optional[str] = "GOAT Routing API"
     CACHE_DIR: str = "/app/src/cache"
 
+    STREET_NETWORK_EDGE_DEFAULT_LAYER_PROJECT_ID = 36126
+    STREET_NETWORK_NODE_DEFAULT_LAYER_PROJECT_ID = 37319
+
     NETWORK_REGION_TABLE = "basic.geofence_active_mobility"
 
     CATCHMENT_AREA_CAR_BUFFER_DEFAULT_SPEED = 80  # km/h
+    CATCHMENT_AREA_HOLE_THRESHOLD_SQM = 10000  # 100m x 100m
+
+    DATA_INSERT_BATCH_SIZE = 800
 
     CELERY_BROKER_URL: Optional[str] = "pyamqp://guest@rabbitmq//"
     REDIS_HOST: Optional[str] = "redis"

src/core/street_network/street_network_cache.py

@@ -0,0 +1,146 @@
+import os
+
+import polars as pl
+from polars import DataFrame
+
+from src.core.config import settings
+
+
+class StreetNetworkCache:
+    def __init__(self):
+        """Initialize the cache directory if it does not exist."""
+
+        if not os.path.exists(settings.CACHE_DIR):
+            os.makedirs(settings.CACHE_DIR)
+
+    def _get_edge_cache_file_name(
+        self,
+        edge_layer_project_id: int,
+        h3_short: str,
+    ):
+        """Get edge cache file path for the specified H3_3 cell."""
+
+        return os.path.join(
+            settings.CACHE_DIR,
+            f"{edge_layer_project_id}_{h3_short}_edge.parquet",
+        )
+
+    def _get_node_cache_file_name(
+        self,
+        node_layer_project_id: int,
+        h3_short: str,
+    ):
+        """Get node cache file path for the specified H3_3 cell."""
+
+        return os.path.join(
+            settings.CACHE_DIR,
+            f"{node_layer_project_id}_{h3_short}_node.parquet",
+        )
+
+    def edge_cache_exists(self, edge_layer_project_id: int, h3_short: str):
+        """Check if edge data for the specified H3_3 cell is cached."""
+
+        edge_cache_file = self._get_edge_cache_file_name(
+            edge_layer_project_id, h3_short
+        )
+        return os.path.exists(edge_cache_file)
+
+    def node_cache_exists(self, node_layer_project_id: int, h3_short: str):
+        """Check if node data for the specified H3_3 cell is cached."""
+
+        node_cache_file = self._get_node_cache_file_name(
+            node_layer_project_id, h3_short
+        )
+        return os.path.exists(node_cache_file)
+
+    def read_edge_cache(
+        self,
+        edge_layer_project_id: int,
+        h3_short: str,
+    ):
+        """Read edge data for the specified H3_3 cell from cache."""
+
+        edge_df: DataFrame = None
+
+        edge_cache_file = self._get_edge_cache_file_name(
+            edge_layer_project_id, h3_short
+        )
+
+        try:
+            with open(edge_cache_file, "rb") as file:
+                edge_df = pl.read_parquet(file)
+        except Exception:
+            raise ValueError(
+                f"Failed to read edge data for H3_3 cell {h3_short} from cache."
+            )
+
+        return edge_df
+
+    def read_node_cache(
+        self,
+        node_layer_project_id: int,
+        h3_short: str,
+    ):
+        """Read node data for the specified H3_3 cell from cache."""
+
+        node_df: DataFrame = None
+
+        node_cache_file = self._get_node_cache_file_name(
+            node_layer_project_id, h3_short
+        )
+
+        try:
+            with open(node_cache_file, "rb") as file:
+                node_df = pl.read_parquet(file)
+        except Exception:
+            raise ValueError(
+                f"Failed to read node data for H3_3 cell {h3_short} from cache."
+            )
+
+        return node_df
+
+    def write_edge_cache(
+        self,
+        edge_layer_project_id: int,
+        h3_short: str,
+        edge_df: DataFrame,
+    ):
+        """Write edge data for the specified H3_3 cell into cache."""
+
+        edge_cache_file = self._get_edge_cache_file_name(
+            edge_layer_project_id, h3_short
+        )
+
+        try:
+            with open(edge_cache_file, "wb") as file:
+                edge_df.write_parquet(file)
+        except Exception:
+            # Clean up cache file if writing fails
+            if os.path.exists(edge_cache_file):
+                os.remove(edge_cache_file)
+            raise RuntimeError(
+                f"Failed to write edge data for H3_3 cell {h3_short} into cache."
+            )
+
+    def write_node_cache(
+        self,
+        node_layer_project_id: int,
+        h3_short: str,
+        node_df: DataFrame,
+    ):
+        """Write node data for the specified H3_3 cell into cache."""
+
+        node_cache_file = self._get_node_cache_file_name(
+            node_layer_project_id, h3_short
+        )
+
+        try:
+            with open(node_cache_file, "wb") as file:
+                node_df.write_parquet(file)
+        except Exception:
+            # Clean up cache file if writing fails
+            if os.path.exists(node_cache_file):
+                os.remove(node_cache_file)
+            raise RuntimeError(
+                f"Failed to write node data for H3_3 cell {h3_short} into cache."
+            )