boundaries plz

Author: cameronbrill

core/map/main.py

@@ -1,7 +1,6 @@
 from enum import Enum
 from pathlib import Path
 import re
-from typing import Callable
 
 import geopandas as gpd
 import httpx
@@ -50,8 +49,14 @@ def fetch_geojson_data(url: str, timeout: float = 30.0) -> str:
         httpx.RequestError: For network errors
         httpx.HTTPStatusError: For HTTP errors
     """
+    headers = {}
+
+    # Add User-Agent for OpenStreetMap/Nominatim requests
+    if "nominatim.openstreetmap.org" in url:
+        headers["User-Agent"] = "jet-lag-munich/0.1.0 (https://github.com/cameronbrill/jet-lag-munich)"
+
     with httpx.Client() as client:
-        response = client.get(url, timeout=timeout)
+        response = client.get(url, timeout=timeout, headers=headers)
         response.raise_for_status()
         return response.text
 
@@ -70,6 +75,52 @@ def separate_geometries(gdf: gpd.GeoDataFrame) -> tuple[gpd.GeoDataFrame, gpd.Ge
     return points_gdf, lines_gdf
 
 
+def extract_boundary_polygon(boundary_gdf: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
+    """Extract boundary polygon from polygon/multipolygon geometries.
+
+    Args:
+        boundary_gdf: GeoDataFrame containing polygon geometries
+
+    Returns:
+        GeoDataFrame with Polygon geometry for Google My Maps area tinting
+    """
+
+    logger = get_logger(__name__)
+    boundary_lines = []
+
+    for idx, row in boundary_gdf.iterrows():
+        geom = row.geometry
+
+        if geom.geom_type == "Polygon":
+            # Keep the polygon as-is for Google My Maps tinting
+            boundary_polygon = geom
+        elif geom.geom_type == "MultiPolygon":
+            # Find the largest polygon and keep it as a polygon
+            boundary_polygon = max(geom.geoms, key=lambda p: p.area)
+        else:
+            # Skip non-polygon geometries
+            logger.debug("Skipping non-polygon geometry", geom_type=geom.geom_type)
+            continue
+
+        # Create new row with boundary polygon
+        new_row = row.copy()
+        new_row.geometry = boundary_polygon
+        new_row["name"] = "Munich Boundary"
+        boundary_lines.append(new_row)
+
+        logger.info(
+            "Extracted boundary polygon",
+            geom_type=geom.geom_type,
+            boundary_points=len(boundary_polygon.exterior.coords),
+            area=boundary_polygon.area
+        )
+
+    if boundary_lines:
+        return gpd.GeoDataFrame(boundary_lines, crs=boundary_gdf.crs)
+    # Return empty GeoDataFrame with same structure
+    return gpd.GeoDataFrame(columns=boundary_gdf.columns, crs=boundary_gdf.crs)
+
+
 def create_stations_csv(points_gdf: gpd.GeoDataFrame, endpoint_name: str) -> pd.DataFrame:
     """Create CSV DataFrame for stations with Google My Maps compatible columns.
 
@@ -97,9 +148,12 @@ def create_stations_csv(points_gdf: gpd.GeoDataFrame, endpoint_name: str) -> pd.
 
     if "station_label" in stations_df.columns:
         # Use station_label value directly, fallback to rail-type specific description
-        station_label_application_func: Callable[[str], str] = lambda x: str(x).strip() if pd.notna(x) and str(x).strip() != "" else fallback_description
+        def _station_label_application_func(x: str) -> str:
+            if pd.notna(x) and str(x).strip() != "":
+                return str(x).strip()
+            return fallback_description
         stations_df["Description"] = stations_df["station_label"].apply( # pyright: ignore[reportUnknownMemberType]
-            station_label_application_func
+            _station_label_application_func
         )
     else:
         stations_df["Description"] = fallback_description
@@ -282,7 +336,7 @@ class MunichGeoJson(str, Enum):
     SUBWAY_LIGHTRAIL = "https://loom.cs.uni-freiburg.de/components/subway-lightrail/13/component-220.json"
     TRAM = "https://loom.cs.uni-freiburg.de/components/tram/13/component-176.json"
     COMMUTER_RAIL = "https://loom.cs.uni-freiburg.de/components/rail-commuter/13/component-78.json"
-    BOUNDARY = "https://nominatim.openstreetmap.org/search.php?q=Munich&polygon_geojson=1&format=geojson"
+    BOUNDARY = "https://nominatim.openstreetmap.org/search.php?q=Munich&polygon_geojson=1&format=geojson&countrycodes=de"
 
 
 def main() -> None:
@@ -343,7 +397,39 @@ def main() -> None:
                 gdf = gdf.to_crs("EPSG:4326")
                 logger.info("Converted CRS to WGS84", endpoint=endpoint.name)
 
-            # Separate geometries
+            # Handle boundary data differently
+            if endpoint.name == "BOUNDARY":
+                # Extract boundary polygon from polygon/multipolygon
+                boundary_polygon_gdf = extract_boundary_polygon(gdf)
+
+                if len(boundary_polygon_gdf) > 0:
+                    # Create CSV for boundary (will use WKT POLYGON format for Google My Maps tinting)
+                    boundary_csv = create_lines_csv(boundary_polygon_gdf)
+                    boundary_csv_file = output_dir / "munich_boundary.csv"
+                    boundary_csv.to_csv(boundary_csv_file, index=False)
+
+                    # Create KML for boundary
+                    boundary_kml = output_dir / "munich_boundary.kml"
+                    create_simple_kml(boundary_polygon_gdf, "munich_boundary", boundary_kml)
+
+                    logger.info(
+                        "Successfully converted boundary to CSV and KML",
+                        endpoint=endpoint.name,
+                        csv_file=str(boundary_csv_file),
+                        csv_size_bytes=boundary_csv_file.stat().st_size,
+                        kml_file=str(boundary_kml),
+                        kml_size_bytes=boundary_kml.stat().st_size,
+                        boundary_features=len(boundary_polygon_gdf),
+                    )
+                else:
+                    logger.warning("No boundary data found", endpoint=endpoint.name)
+
+                # Clean up and continue to next endpoint
+                temp_geojson.unlink()
+                logger.debug("Cleaned up temporary file", temp_file=str(temp_geojson))
+                continue
+
+            # Separate geometries for transit data
             points_gdf, lines_gdf = separate_geometries(gdf)
 
             # Google My Maps limits: max 2,000 rows, 5MB file size

tests/map/test_main.py

@@ -15,6 +15,7 @@
     create_lines_csv,
     create_simple_kml,
     create_stations_csv,
+    extract_boundary_polygon,
     extract_line_name,
     fetch_geojson_data,
     main,
@@ -86,7 +87,7 @@ def test_fetches_data_successfully(self, mock_client_class):
         result = fetch_geojson_data("https://example.com/data.json")
 
         assert result == '{"type": "FeatureCollection"}'
-        mock_client.get.assert_called_once_with("https://example.com/data.json", timeout=30.0)
+        mock_client.get.assert_called_once_with("https://example.com/data.json", timeout=30.0, headers={})
 
     @patch("core.map.main.httpx.Client")
     def test_raises_request_error_on_network_failure(self, mock_client_class):
@@ -203,6 +204,79 @@ def test_includes_description_with_station_names(self):
         assert result.iloc[1]["Description"] == "Unknown Subway Station"  # Rail-type specific fallback
 
 
+class TestExtractBoundaryLine:
+    """Test extraction of boundary line from polygon/multipolygon geometries."""
+
+    def test_preserves_polygon_geometry_for_google_maps_tinting(self):
+        """Should preserve polygon geometry to enable Google My Maps area tinting."""
+        from shapely.geometry import Polygon
+
+        # Create a simple polygon (square)
+        polygon = Polygon([(11.0, 48.0), (11.1, 48.0), (11.1, 48.1), (11.0, 48.1), (11.0, 48.0)])
+        data = {
+            "geometry": [polygon],
+            "name": ["Munich"]
+        }
+        boundary_gdf = gpd.GeoDataFrame(data, crs="EPSG:4326")
+
+        result = extract_boundary_polygon(boundary_gdf)
+
+        # Should return a GeoDataFrame with Polygon geometry (not LineString)
+        assert len(result) == 1
+        assert result.iloc[0].geometry.geom_type == "Polygon"
+        assert "name" in result.columns
+        assert result.iloc[0]["name"] == "Munich Boundary"
+
+    def test_handles_multipolygon_by_taking_largest(self):
+        """Should extract boundary from the largest polygon in a MultiPolygon."""
+        from shapely.geometry import MultiPolygon, Polygon
+
+        # Create MultiPolygon with different sized polygons
+        small_poly = Polygon([(11.0, 48.0), (11.01, 48.0), (11.01, 48.01), (11.0, 48.01), (11.0, 48.0)])
+        large_poly = Polygon([(11.0, 48.0), (11.2, 48.0), (11.2, 48.2), (11.0, 48.2), (11.0, 48.0)])
+        multipolygon = MultiPolygon([small_poly, large_poly])
+
+        data = {
+            "geometry": [multipolygon],
+            "name": ["Munich"]
+        }
+        boundary_gdf = gpd.GeoDataFrame(data, crs="EPSG:4326")
+
+        result = extract_boundary_polygon(boundary_gdf)
+
+        # Should extract the larger polygon
+        assert len(result) == 1
+        assert result.iloc[0].geometry.geom_type == "Polygon"
+        # The boundary should have more area (from the larger polygon)
+        assert result.iloc[0].geometry.area > 0.01
+
+    def test_creates_polygon_wkt_for_google_maps_tinting(self):
+        """Should create POLYGON WKT format for Google My Maps area tinting."""
+        from shapely.geometry import Polygon
+
+        # Create a simple polygon
+        polygon = Polygon([(11.0, 48.0), (11.1, 48.0), (11.1, 48.1), (11.0, 48.1), (11.0, 48.0)])
+        data = {
+            "geometry": [polygon],
+            "name": ["Munich"]
+        }
+        boundary_gdf = gpd.GeoDataFrame(data, crs="EPSG:4326")
+
+        result = extract_boundary_polygon(boundary_gdf)
+
+        # Should preserve polygon for area tinting
+        assert len(result) == 1
+        assert result.iloc[0].geometry.geom_type == "Polygon"
+
+        # When converted to CSV, should create POLYGON WKT
+        boundary_csv = create_lines_csv(result)
+        wkt_value = boundary_csv.iloc[0]["WKT"]
+        assert wkt_value.startswith("POLYGON")
+        # The name will be "Unknown Line" since boundary polygons don't have dbg_lines
+        # but the important thing is we get POLYGON WKT format
+        assert boundary_csv.iloc[0]["name"] == "Unknown Line"
+
+
 class TestCreateSimpleKml:
     """Test creation of simplified KML files with Google My Maps compatible attributes."""
 
@@ -379,5 +453,5 @@ def test_main_processes_all_endpoints(self, mock_fetch):
             # This should not raise any exceptions
             main()
 
-        # Should have called fetch for all endpoints
-        assert mock_fetch.call_count == 3
+        # Should have called fetch for all endpoints (including BOUNDARY)
+        assert mock_fetch.call_count == 4