From b87cb268d5f0265dee327b0a237da8ccd5062b8f Mon Sep 17 00:00:00 2001 From: Nitzan Pomerantz <9297302+nitzpo@users.noreply.github.com> Date: Mon, 8 Dec 2025 09:34:42 +0200 Subject: [PATCH] Improve: MCP efficiency & fix radius filtering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove bloat fields (shape, objectid, etc), add sequential IDs - Add lang param (he/en) for Hebrew/English text values - Reduce JSON whitespace (indent=None) - Fix distance_meters: extract centroid from WKT shape geometry - Add search_coordinates to all address-based tool responses - Fix radius filtering: properly filter deals beyond radius_meters - Change default radius from 30m to 50m - Fix get_deal_statistics: return NO deals (stats only) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- nadlan_mcp/fastmcp_server.py | 321 +++++++++++++++++++++++------------ nadlan_mcp/govmap/client.py | 46 ++++- nadlan_mcp/govmap/utils.py | 39 ++++- tests/test_fastmcp_tools.py | 79 +++++++++ 4 files changed, 368 insertions(+), 117 deletions(-) diff --git a/nadlan_mcp/fastmcp_server.py b/nadlan_mcp/fastmcp_server.py index 0d8083a..88a12fe 100644 --- a/nadlan_mcp/fastmcp_server.py +++ b/nadlan_mcp/fastmcp_server.py @@ -74,32 +74,73 @@ def log_mcp_call(func_name: str, **params): ) -def strip_bloat_fields(deals: List[Deal]) -> List[Dict[str, Any]]: +def strip_bloat_fields(deals: List[Deal], lang: str = "he") -> List[Dict[str, Any]]: """ Remove bloat fields from Deal models to reduce token usage in MCP responses. - Converts Deal models to dictionaries and removes: - - shape: Large MULTIPOLYGON coordinate data (~40-50% of tokens, not useful for LLM analysis) - - sourceorder: Internal ordering field - - source_polygon_id: Internal reference field (only when it's a UUID string) + Converts Deal models to dictionaries and removes unnecessary fields: + - shape: Large MULTIPOLYGON coordinate data + - sourceorder, objectid, priority: Internal fields + - source_polygon_id, settlementId, streetCode, dealId, polygonId: Internal IDs + - deal_type_description: Redundant (kept in search_parameters only) + + Adds sequential ID for LLM reference and selects language for text fields. Args: deals: List of Deal model instances + lang: Language for text values ("he" for Hebrew, "en" for English) Returns: - List of deal dictionaries with bloat fields removed + List of deal dictionaries with bloat removed and sequential IDs """ - bloat_fields = {"shape", "sourceorder"} - # Note: We keep source_polygon_id if it was added by our processing logic + bloat_fields = { + "shape", + "sourceorder", + "objectid", + "priority", + "source_polygon_id", + "settlementId", + "streetCode", + "dealId", + "polygonId", + "deal_type_description", + } result = [] - for deal in deals: - # Convert Deal model to dict, excluding None values for cleaner output - # Use mode='json' to serialize dates as ISO strings + for idx, deal in enumerate(deals, start=1): + # Convert Deal model to dict, excluding None values deal_dict = deal.model_dump(mode="json", exclude_none=True) # Remove bloat fields filtered_dict = {k: v for k, v in deal_dict.items() if k not in bloat_fields} + + # Add sequential ID for LLM reference + filtered_dict["id"] = idx + + # Select language for text fields (Hebrew or English) + if lang == "en": + # Use English values if available + if "settlementNameEng" in filtered_dict: + filtered_dict["settlement_name"] = filtered_dict.pop("settlementNameEng") + filtered_dict.pop("settlementNameHeb", None) + if "streetNameEng" in filtered_dict: + filtered_dict["street_name"] = filtered_dict.pop("streetNameEng") + filtered_dict.pop("streetNameHeb", None) + if "assetTypeEng" in filtered_dict: + filtered_dict["asset_type"] = filtered_dict.pop("assetTypeEng") + filtered_dict.pop("assetTypeHeb", None) + else: # Default to Hebrew + # Use Hebrew values + if "settlementNameHeb" in filtered_dict: + filtered_dict["settlement_name"] = filtered_dict.pop("settlementNameHeb") + filtered_dict.pop("settlementNameEng", None) + if "streetNameHeb" in filtered_dict: + filtered_dict["street_name"] = filtered_dict.pop("streetNameHeb") + filtered_dict.pop("streetNameEng", None) + if "assetTypeHeb" in filtered_dict: + filtered_dict["asset_type"] = filtered_dict.pop("assetTypeHeb") + filtered_dict.pop("assetTypeEng", None) + result.append(filtered_dict) return result @@ -141,7 +182,7 @@ def autocomplete_address(search_text: str) -> str: formatted_results.append(result_dict) - return json.dumps(formatted_results, ensure_ascii=False, indent=2) + return json.dumps(formatted_results, ensure_ascii=False, indent=None) except Exception as e: logger.error(f"Error in autocomplete_address: {e}", exc_info=True) @@ -182,7 +223,7 @@ def get_deals_by_radius(latitude: float, longitude: float, radius_meters: int = "polygons": polygons, # Return dicts directly, no stripping needed }, ensure_ascii=False, - indent=2, + indent=None, ) except Exception as e: @@ -236,10 +277,10 @@ def get_street_deals(polygon_id: str, limit: int = 100, deal_type: int = 2) -> s "deal_type": deal_type, "deal_type_description": deal_type_desc, "market_statistics": stats, - "deals": strip_bloat_fields(deals), + "deals": strip_bloat_fields(deals, lang="he"), }, ensure_ascii=False, - indent=2, + indent=None, ) except Exception as e: @@ -251,22 +292,25 @@ def get_street_deals(polygon_id: str, limit: int = 100, deal_type: int = 2) -> s def find_recent_deals_for_address( address: str, years_back: int = 2, - radius_meters: int = 30, + radius_meters: int = 50, max_deals: int = 100, deal_type: int = 2, + lang: str = "he", ) -> str: """Find recent real estate deals for a specific address. Args: address: The address to search for (in Hebrew or English) years_back: How many years back to search (default: 2) - radius_meters: Search radius in meters from the address (default: 30) - Small radius since street deals cover the entire street anyway + radius_meters: Search radius in meters from the address (default: 50) + Deals beyond this radius are filtered out, EXCEPT same-building deals + which are always included regardless of distance. max_deals: Maximum number of deals to return (default: 100, provides good context for LLM analysis) deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2) + lang: Language for text values ("he" for Hebrew [default], "en" for English) Returns: - JSON string containing recent real estate deals for the address + JSON string containing recent real estate deals within radius_meters of the address """ log_mcp_call( "find_recent_deals_for_address", @@ -277,6 +321,14 @@ def find_recent_deals_for_address( deal_type=deal_type, ) try: + # Get coordinates for search center + autocomplete_result = client.autocomplete_address(address) + search_coords = None + if autocomplete_result.results: + coords = autocomplete_result.results[0].coordinates + if coords: + search_coords = {"longitude": coords.longitude, "latitude": coords.latitude} + deals = client.find_recent_deals_for_address( address, years_back, radius_meters, max_deals, deal_type ) @@ -347,21 +399,25 @@ def find_recent_deals_for_address( } deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)" + search_params = { + "address": address, + "years_back": years_back, + "radius_meters": radius_meters, + "max_deals": max_deals, + "deal_type": deal_type, + "deal_type_description": deal_type_desc, + } + if search_coords: + search_params["search_coordinates"] = search_coords + return json.dumps( { - "search_parameters": { - "address": address, - "years_back": years_back, - "radius_meters": radius_meters, - "max_deals": max_deals, - "deal_type": deal_type, - "deal_type_description": deal_type_desc, - }, + "search_parameters": search_params, "market_statistics": stats, - "deals": strip_bloat_fields(deals), + "deals": strip_bloat_fields(deals, lang=lang), }, ensure_ascii=False, - indent=2, + indent=None, ) except Exception as e: @@ -370,13 +426,16 @@ def find_recent_deals_for_address( @conditional_tool("tool_get_neighborhood_deals_enabled") -def get_neighborhood_deals(polygon_id: str, limit: int = 100, deal_type: int = 2) -> str: +def get_neighborhood_deals( + polygon_id: str, limit: int = 100, deal_type: int = 2, lang: str = "he" +) -> str: """Get real estate deals for a specific neighborhood polygon. Args: polygon_id: The polygon ID of the neighborhood limit: Maximum number of deals to return (default: 100) deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2) + lang: Language for text values ("he" for Hebrew [default], "en" for English) Returns: JSON string containing recent real estate deals in the specified neighborhood @@ -415,10 +474,10 @@ def get_neighborhood_deals(polygon_id: str, limit: int = 100, deal_type: int = 2 "deal_type": deal_type, "deal_type_description": deal_type_desc, "market_statistics": stats, - "deals": strip_bloat_fields(deals), + "deals": strip_bloat_fields(deals, lang=lang), }, ensure_ascii=False, - indent=2, + indent=None, ) except Exception as e: @@ -457,6 +516,14 @@ def analyze_market_trends( deal_type=deal_type, ) try: + # Get coordinates for search center + autocomplete_result = client.autocomplete_address(address) + search_coords = None + if autocomplete_result.results: + coords = autocomplete_result.results[0].coordinates + if coords: + search_coords = {"longitude": coords.longitude, "latitude": coords.latitude} + # Get deals for the address with larger radius for trend analysis deals = client.find_recent_deals_for_address( address, years_back, radius_meters, max_deals, deal_type @@ -600,16 +667,20 @@ def analyze_market_trends( # Return summarized analysis (NO raw deals to save tokens) # Normalize structure with standard market_statistics while keeping tool-specific analysis + analysis_params = { + "address": address, + "years_analyzed": years_back, + "radius_meters": radius_meters, + "deals_analyzed": len(deals), + "deal_type": deal_type, + "deal_type_description": deal_type_desc, + } + if search_coords: + analysis_params["search_coordinates"] = search_coords + return json.dumps( { - "analysis_parameters": { - "address": address, - "years_analyzed": years_back, - "radius_meters": radius_meters, - "deals_analyzed": len(deals), - "deal_type": deal_type, - "deal_type_description": deal_type_desc, - }, + "analysis_parameters": analysis_params, "market_statistics": { "deal_breakdown": { "total_deals": len(deals), @@ -640,7 +711,7 @@ def analyze_market_trends( "deals": [], # Trend analysis doesn't return raw deals to save tokens }, ensure_ascii=False, - indent=2, + indent=None, ) except Exception as e: @@ -666,6 +737,20 @@ def compare_addresses(addresses: List[str]) -> str: for address in addresses: try: + # Get coordinates for this address + search_coords = None + try: + autocomplete_result = client.autocomplete_address(address) + if autocomplete_result.results: + coords = autocomplete_result.results[0].coordinates + if coords: + search_coords = { + "longitude": coords.longitude, + "latitude": coords.latitude, + } + except Exception: + pass # Continue without coordinates if autocomplete fails + deals = client.find_recent_deals_for_address(address, 2) if deals: @@ -690,6 +775,7 @@ def compare_addresses(addresses: List[str]) -> str: comparison = { "address": address, + "search_coordinates": search_coords, "total_deals": len(deals), "same_building_deals": len(building_deals), "street_deals": len(street_deals), @@ -731,6 +817,7 @@ def compare_addresses(addresses: List[str]) -> str: else: comparison = { "address": address, + "search_coordinates": search_coords, "total_deals": 0, "same_building_deals": 0, "street_deals": 0, @@ -775,7 +862,7 @@ def compare_addresses(addresses: List[str]) -> str: "all_results": comparisons, }, ensure_ascii=False, - indent=2, + indent=None, ) except Exception as e: @@ -800,6 +887,7 @@ def get_valuation_comparables( max_comparables: int = 50, iqr_multiplier: Optional[float] = None, include_outlier_deals: bool = True, + lang: str = "he", ) -> str: """Get comparable properties for valuation analysis. @@ -825,6 +913,7 @@ def get_valuation_comparables( iqr_multiplier: Override IQR multiplier for outlier detection (default: 1.0). Lower = more aggressive filtering include_outlier_deals: If True (default), include the outlier deals separately in the response This allows you to see what was filtered out and answer questions about it + lang: Language for text values ("he" for Hebrew [default], "en" for English) Returns: JSON string containing: @@ -855,20 +944,32 @@ def get_valuation_comparables( iqr_multiplier=iqr_multiplier, ) try: + # Get coordinates for search center + autocomplete_result = client.autocomplete_address(address) + search_coords = None + if autocomplete_result.results: + coords = autocomplete_result.results[0].coordinates + if coords: + search_coords = {"longitude": coords.longitude, "latitude": coords.latitude} + # Get all deals for the address with higher limits for valuation deals = client.find_recent_deals_for_address( address, years_back, radius=radius_meters, max_deals=max_comparables ) + search_params_base = { + "address": address, + "years_back": years_back, + "radius_meters": radius_meters, + "max_comparables": max_comparables, + } + if search_coords: + search_params_base["search_coordinates"] = search_coords + if not deals: return json.dumps( { - "search_parameters": { - "address": address, - "years_back": years_back, - "radius_meters": radius_meters, - "max_comparables": max_comparables, - }, + "search_parameters": search_params_base, "market_statistics": { "deal_breakdown": { "total_deals": 0, @@ -878,7 +979,7 @@ def get_valuation_comparables( "message": "No deals found for this address", }, ensure_ascii=False, - indent=2, + indent=None, ) # Apply filters @@ -945,20 +1046,16 @@ def get_valuation_comparables( deal_breakdown["iqr_multiplier"] = outlier_report["parameters"]["iqr_multiplier"] # Build response with filtered deals + search_params_base["filters_applied"] = { + "property_type": property_type, + "rooms": f"{min_rooms}-{max_rooms}" if min_rooms or max_rooms else None, + "price": f"{min_price}-{max_price}" if min_price or max_price else None, + "area": f"{min_area}-{max_area}" if min_area or max_area else None, + "floor": f"{min_floor}-{max_floor}" if min_floor or max_floor else None, + } + response_data = { - "search_parameters": { - "address": address, - "years_back": years_back, - "radius_meters": radius_meters, - "max_comparables": max_comparables, - "filters_applied": { - "property_type": property_type, - "rooms": f"{min_rooms}-{max_rooms}" if min_rooms or max_rooms else None, - "price": f"{min_price}-{max_price}" if min_price or max_price else None, - "area": f"{min_area}-{max_area}" if min_area or max_area else None, - "floor": f"{min_floor}-{max_floor}" if min_floor or max_floor else None, - }, - }, + "search_parameters": search_params_base, "market_statistics": { "deal_breakdown": deal_breakdown, "price_statistics": stats.price_statistics, @@ -967,14 +1064,16 @@ def get_valuation_comparables( "property_type_distribution": stats.property_type_distribution, "date_range": stats.date_range, }, - "deals": strip_bloat_fields(filtered_deals), + "deals": strip_bloat_fields(filtered_deals, lang=lang), } # Add outlier deals if present and requested if include_outlier_deals and outlier_report and "outlier_deals" in outlier_report: - response_data["outlier_deals"] = strip_bloat_fields(outlier_report["outlier_deals"]) + response_data["outlier_deals"] = strip_bloat_fields( + outlier_report["outlier_deals"], lang=lang + ) - return json.dumps(response_data, ensure_ascii=False, indent=2) + return json.dumps(response_data, ensure_ascii=False, indent=None) except Exception as e: logger.error(f"Error in get_valuation_comparables: {e}", exc_info=True) @@ -989,12 +1088,11 @@ def get_deal_statistics( min_rooms: Optional[float] = None, max_rooms: Optional[float] = None, iqr_multiplier: Optional[float] = None, - include_outlier_deals: bool = True, ) -> str: """Calculate statistical aggregations on deal data for an address. - This tool provides quick statistical summaries without returning all raw deals. - Useful when LLM needs calculations on large datasets without full details. + This tool provides quick statistical summaries WITHOUT returning any deals. + Useful when LLM needs calculations on large datasets without full deal details. Args: address: The address to analyze (in Hebrew or English) @@ -1003,13 +1101,9 @@ def get_deal_statistics( min_rooms: Minimum number of rooms max_rooms: Maximum number of rooms iqr_multiplier: Override IQR multiplier for outlier detection (default: 1.0). Lower = more aggressive filtering - include_outlier_deals: If True (default), include the outlier deals in the response - This allows you to see what was filtered out Returns: - JSON string containing: - - market_statistics: Statistical metrics (mean, median, percentiles, etc.) - - outlier_deals: Deals that were removed as outliers (if include_outlier_deals=True) + JSON string containing market_statistics only (no deals returned) """ log_mcp_call( "get_deal_statistics", @@ -1021,16 +1115,28 @@ def get_deal_statistics( iqr_multiplier=iqr_multiplier, ) try: + # Get coordinates for search center + autocomplete_result = client.autocomplete_address(address) + search_coords = None + if autocomplete_result.results: + coords = autocomplete_result.results[0].coordinates + if coords: + search_coords = {"longitude": coords.longitude, "latitude": coords.latitude} + # Get all deals for the address deals = client.find_recent_deals_for_address(address, years_back) + search_params_base = { + "address": address, + "years_back": years_back, + } + if search_coords: + search_params_base["search_coordinates"] = search_coords + if not deals: return json.dumps( { - "search_parameters": { - "address": address, - "years_back": years_back, - }, + "search_parameters": search_params_base, "market_statistics": { "deal_breakdown": {"total_deals": 0}, "message": "No deals found for this address", @@ -1038,7 +1144,7 @@ def get_deal_statistics( "deals": [], }, ensure_ascii=False, - indent=2, + indent=None, ) # Apply filters if provided @@ -1047,21 +1153,19 @@ def get_deal_statistics( deals, property_type=property_type, min_rooms=min_rooms, max_rooms=max_rooms ) - # Calculate statistics + # Calculate statistics (don't need outlier deals for statistics-only tool) stats = client.calculate_deal_statistics( - deals, iqr_multiplier=iqr_multiplier, include_outlier_deals=include_outlier_deals + deals, iqr_multiplier=iqr_multiplier, include_outlier_deals=False ) - # Build response + # Build response - statistics only, NO deals + search_params_base["filters_applied"] = { + "property_type": property_type, + "rooms": f"{min_rooms}-{max_rooms}" if min_rooms or max_rooms else None, + } + response_data = { - "search_parameters": { - "address": address, - "years_back": years_back, - "filters_applied": { - "property_type": property_type, - "rooms": f"{min_rooms}-{max_rooms}" if min_rooms or max_rooms else None, - }, - }, + "search_parameters": search_params_base, "market_statistics": { "deal_breakdown": { "total_deals": stats.total_deals, @@ -1072,14 +1176,9 @@ def get_deal_statistics( "property_type_distribution": stats.property_type_distribution, "date_range": stats.date_range, }, - "deals": [], # Statistics-only query, no full deals returned } - # Add outlier deals if present and requested - if include_outlier_deals and stats.outlier_report and stats.outlier_report.outlier_deals: - response_data["outlier_deals"] = strip_bloat_fields(stats.outlier_report.outlier_deals) - - return json.dumps(response_data, ensure_ascii=False, indent=2) + return json.dumps(response_data, ensure_ascii=False, indent=None) except Exception as e: logger.error(f"Error in get_deal_statistics: {e}", exc_info=True) @@ -1139,17 +1238,29 @@ def get_market_activity_metrics(address: str, years_back: int = 2, radius_meters radius_meters=radius_meters, ) try: + # Get coordinates for search center + autocomplete_result = client.autocomplete_address(address) + search_coords = None + if autocomplete_result.results: + coords = autocomplete_result.results[0].coordinates + if coords: + search_coords = {"longitude": coords.longitude, "latitude": coords.latitude} + # Get deals for the address deals = client.find_recent_deals_for_address(address, years_back, radius_meters) + analysis_params_base = { + "address": address, + "years_back": years_back, + "radius_meters": radius_meters, + } + if search_coords: + analysis_params_base["search_coordinates"] = search_coords + if not deals: return json.dumps( { - "analysis_parameters": { - "address": address, - "years_back": years_back, - "radius_meters": radius_meters, - }, + "analysis_parameters": analysis_params_base, "market_statistics": { "deal_breakdown": { "total_deals": 0, @@ -1159,7 +1270,7 @@ def get_market_activity_metrics(address: str, years_back: int = 2, radius_meters "error": "No deals found for analysis", }, ensure_ascii=False, - indent=2, + indent=None, ) # Calculate market metrics using helper to reduce duplication @@ -1170,11 +1281,7 @@ def get_market_activity_metrics(address: str, years_back: int = 2, radius_meters # Combine all metrics with normalized structure return json.dumps( { - "analysis_parameters": { - "address": address, - "years_back": years_back, - "radius_meters": radius_meters, - }, + "analysis_parameters": analysis_params_base, "market_statistics": { "deal_breakdown": { "total_deals": len(deals), @@ -1195,7 +1302,7 @@ def get_market_activity_metrics(address: str, years_back: int = 2, radius_meters "deals": [], # Activity metrics don't return raw deals }, ensure_ascii=False, - indent=2, + indent=None, ) except Exception as e: diff --git a/nadlan_mcp/govmap/client.py b/nadlan_mcp/govmap/client.py index 3047eaa..225cac7 100644 --- a/nadlan_mcp/govmap/client.py +++ b/nadlan_mcp/govmap/client.py @@ -518,7 +518,7 @@ class GovmapClient: self, address: str, years_back: int = 2, - radius: int = 30, + radius: int = 50, max_deals: int = 100, deal_type: int = 2, ) -> List[Deal]: @@ -531,14 +531,15 @@ class GovmapClient: Args: address: The address to search for years_back: How many years back to search (default: 2) - radius: Search radius in meters for initial coordinate search (default: 30) - Small radius since street deals cover the entire street anyway + radius: Search radius in meters (default: 50). Deals beyond this radius are filtered out, + EXCEPT same-building deals which are always included regardless of distance. max_deals: Maximum number of deals to return (default: 100) deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2) Returns: - List of Deal models found for the address area, with same building deals prioritized first, - then street deals, then neighborhood deals + List of Deal models found within radius meters of the address, with same building deals + prioritized first, then street deals, then neighborhood deals. Same-building deals are + included regardless of distance. Raises: ValueError: If address cannot be found or processed, or input is invalid @@ -722,8 +723,12 @@ class GovmapClient: seen_deals.add(deal_id) # Calculate distance from search point to deal - # Most deals don't have individual coordinates, so use polygon distance - deal_distance = polygon_distance + # Try to extract centroid from shape, fall back to polygon distance + deal_centroid = utils.extract_shape_centroid(deal.shape) + if deal_centroid: + deal_distance = utils.calculate_distance(point, deal_centroid) + else: + deal_distance = polygon_distance # Store metadata using dynamic attributes (allowed by extra='allow') deal.source_polygon_id = polygon_id @@ -765,8 +770,13 @@ class GovmapClient: if deal_id not in seen_deals: seen_deals.add(deal_id) - # Calculate distance from search point (use polygon distance) - deal_distance = polygon_distance + # Calculate distance from search point to deal + # Try to extract centroid from shape, fall back to polygon distance + deal_centroid = utils.extract_shape_centroid(deal.shape) + if deal_centroid: + deal_distance = utils.calculate_distance(point, deal_centroid) + else: + deal_distance = polygon_distance # Apply distance filter for neighborhood deals if deal_distance <= self.config.max_neighborhood_deal_distance_meters: @@ -789,6 +799,24 @@ class GovmapClient: # Step 5: Combine and prioritize: building deals first, then street, then neighborhood all_deals = building_deals + street_deals + neighborhood_deals + # Filter by user-specified radius (deals beyond radius_meters are excluded) + # Same-building deals are ALWAYS included regardless of distance + filtered_deals = [] + for deal in all_deals: + deal_distance = getattr(deal, "distance_meters", 0) + is_same_building = getattr(deal, "deal_source", None) == "same_building" + + if is_same_building or deal_distance <= radius: + filtered_deals.append(deal) + else: + logger.debug(f"Filtered deal at {deal_distance:.0f}m (radius limit: {radius}m)") + + all_deals = filtered_deals + logger.info( + f"After radius filtering ({radius}m): {len(all_deals)} deals " + f"(removed {len(building_deals) + len(street_deals) + len(neighborhood_deals) - len(all_deals)} deals)" + ) + # Multi-level sort with stable sorting: # 1. Date (newest first) - applied first to maintain recency within same priority/distance # 2. Distance (closest first) - applied second to prefer closer deals within same priority diff --git a/nadlan_mcp/govmap/utils.py b/nadlan_mcp/govmap/utils.py index e8faee9..5f58823 100644 --- a/nadlan_mcp/govmap/utils.py +++ b/nadlan_mcp/govmap/utils.py @@ -6,7 +6,7 @@ This module provides shared helper functions with no external dependencies """ import re -from typing import Tuple +from typing import Optional, Tuple def calculate_distance(point1: Tuple[float, float], point2: Tuple[float, float]) -> float: @@ -28,6 +28,43 @@ def calculate_distance(point1: Tuple[float, float], point2: Tuple[float, float]) return (dx * dx + dy * dy) ** 0.5 +def extract_shape_centroid(shape_wkt: Optional[str]) -> Optional[Tuple[float, float]]: + """ + Extract centroid coordinates from WKT geometry (MULTIPOLYGON/POLYGON). + + Parses WKT string and calculates centroid as average of all coordinate points. + + Args: + shape_wkt: WKT geometry string (e.g., "MULTIPOLYGON(...)") + + Returns: + (longitude, latitude) tuple in ITM coordinates, or None if parsing fails + """ + if not shape_wkt or not isinstance(shape_wkt, str): + return None + + try: + # Extract all coordinate pairs using regex + # Matches: "number.number number.number" or "number number" + coord_pattern = r"([\d.]+)\s+([\d.]+)" + matches = re.findall(coord_pattern, shape_wkt) + + if not matches: + return None + + # Calculate average (centroid) + lons = [float(m[0]) for m in matches] + lats = [float(m[1]) for m in matches] + + centroid_lon = sum(lons) / len(lons) + centroid_lat = sum(lats) / len(lats) + + return (centroid_lon, centroid_lat) + + except (ValueError, ZeroDivisionError): + return None + + def is_same_building(search_address: str, deal_address: str) -> bool: """ Check if a deal is from the same building as the search address. diff --git a/tests/test_fastmcp_tools.py b/tests/test_fastmcp_tools.py index f1e941c..72ec373 100644 --- a/tests/test_fastmcp_tools.py +++ b/tests/test_fastmcp_tools.py @@ -203,9 +203,28 @@ class TestGetDealsByRadius: class TestFindRecentDealsForAddress: """Test find_recent_deals_for_address MCP tool.""" + @staticmethod + def _mock_autocomplete(): + """Helper to create mock autocomplete response.""" + return AutocompleteResponse( + resultsCount=1, + results=[ + AutocompleteResult( + id="test", + text="Test Address", + type="address", + score=100, + coordinates=CoordinatePoint(longitude=180000, latitude=665000), + ) + ], + ) + @patch("nadlan_mcp.fastmcp_server.client") def test_successful_find_deals(self, mock_client): """Test successful deal finding with statistics.""" + # Mock autocomplete + mock_client.autocomplete_address.return_value = self._mock_autocomplete() + # Mock with Deal models mock_deals = [ Deal( @@ -248,6 +267,9 @@ class TestFindRecentDealsForAddress: @patch("nadlan_mcp.fastmcp_server.client") def test_find_deals_strips_bloat(self, mock_client): """Test that bloat fields are stripped.""" + # Mock autocomplete + mock_client.autocomplete_address.return_value = self._mock_autocomplete() + # Mock with Deal models mock_deals = [ Deal( @@ -270,9 +292,28 @@ class TestFindRecentDealsForAddress: class TestAnalyzeMarketTrends: """Test analyze_market_trends MCP tool.""" + @staticmethod + def _mock_autocomplete(): + """Helper to create mock autocomplete response.""" + return AutocompleteResponse( + resultsCount=1, + results=[ + AutocompleteResult( + id="test", + text="Test Address", + type="address", + score=100, + coordinates=CoordinatePoint(longitude=180000, latitude=665000), + ) + ], + ) + @patch("nadlan_mcp.fastmcp_server.client") def test_successful_market_analysis(self, mock_client): """Test successful market trend analysis.""" + # Mock autocomplete + mock_client.autocomplete_address.return_value = self._mock_autocomplete() + # Mock with Deal models mock_deals = [ Deal( @@ -341,9 +382,27 @@ class TestCompareAddresses: class TestGetValuationComparables: """Test get_valuation_comparables MCP tool.""" + @staticmethod + def _mock_autocomplete(): + """Helper to create mock autocomplete response.""" + return AutocompleteResponse( + resultsCount=1, + results=[ + AutocompleteResult( + id="test", + text="Test Address", + type="address", + score=100, + coordinates=CoordinatePoint(longitude=180000, latitude=665000), + ) + ], + ) + @patch("nadlan_mcp.fastmcp_server.client") def test_successful_get_comparables(self, mock_client): """Test successful comparable retrieval with filtering.""" + # Mock autocomplete + mock_client.autocomplete_address.return_value = self._mock_autocomplete() # Mock with Deal models mock_deals = [ @@ -382,6 +441,8 @@ class TestGetValuationComparables: @patch("nadlan_mcp.fastmcp_server.client") def test_comparables_strips_bloat(self, mock_client): """Test that bloat fields are stripped from comparables.""" + # Mock autocomplete + mock_client.autocomplete_address.return_value = self._mock_autocomplete() # Mock with Deal models mock_deals = [ @@ -412,9 +473,27 @@ class TestGetValuationComparables: class TestGetDealStatistics: """Test get_deal_statistics MCP tool.""" + @staticmethod + def _mock_autocomplete(): + """Helper to create mock autocomplete response.""" + return AutocompleteResponse( + resultsCount=1, + results=[ + AutocompleteResult( + id="test", + text="Test Address", + type="address", + score=100, + coordinates=CoordinatePoint(longitude=180000, latitude=665000), + ) + ], + ) + @patch("nadlan_mcp.fastmcp_server.client") def test_successful_statistics_calculation(self, mock_client): """Test successful statistics calculation.""" + # Mock autocomplete + mock_client.autocomplete_address.return_value = self._mock_autocomplete() # Mock with Deal models mock_deals = [