Geospatial Operations

def geoadd(
    self,
    name: KeyT,
    values: Sequence[Union[Tuple[float, float, EncodableT], Tuple[EncodableT, float, float]]],
    nx: bool = False,
    xx: bool = False,
    ch: bool = False
) -> int: ...

def geopos(self, name: KeyT, *values: EncodableT) -> List[Optional[Tuple[float, float]]]: ...

def geodist(
    self,
    name: KeyT,
    place1: EncodableT,
    place2: EncodableT,
    unit: Optional[str] = None
) -> Optional[float]: ...

def geohash(self, name: KeyT, *values: EncodableT) -> List[Optional[str]]: ...

def georadius(
    self,
    name: KeyT,
    longitude: float,
    latitude: float,
    radius: float,
    unit: str = "m",
    withdist: bool = False,
    withcoord: bool = False,
    withhash: bool = False,
    count: Optional[int] = None,
    sort: Optional[str] = None,
    store: Optional[KeyT] = None,
    store_dist: Optional[KeyT] = None
) -> List[Union[bytes, List[Union[bytes, float, Tuple[float, float], int]]]]: ...

def georadiusbymember(
    self,
    name: KeyT,
    member: EncodableT,
    radius: float,
    unit: str = "m",
    withdist: bool = False,
    withcoord: bool = False,
    withhash: bool = False,
    count: Optional[int] = None,
    sort: Optional[str] = None,
    store: Optional[KeyT] = None,
    store_dist: Optional[KeyT] = None
) -> List[Union[bytes, List[Union[bytes, float, Tuple[float, float], int]]]]: ...

def geosearch(
    self,
    name: KeyT,
    member: Optional[EncodableT] = None,
    longitude: Optional[float] = None,
    latitude: Optional[float] = None,
    unit: str = "m",
    radius: Optional[float] = None,
    width: Optional[float] = None,
    height: Optional[float] = None,
    sort: Optional[str] = None,
    count: Optional[int] = None,
    any: bool = False,
    withdist: bool = False,
    withcoord: bool = False,
    withhash: bool = False
) -> List[Union[bytes, List[Union[bytes, float, Tuple[float, float], int]]]]: ...

def geosearchstore(
    self,
    dest: KeyT,
    name: KeyT,
    member: Optional[EncodableT] = None,
    longitude: Optional[float] = None,
    latitude: Optional[float] = None,
    unit: str = "m",
    radius: Optional[float] = None,
    width: Optional[float] = None,
    height: Optional[float] = None,
    sort: Optional[str] = None,
    count: Optional[int] = None,
    any: bool = False,
    storedist: bool = False
) -> int: ...

import fakeredis

client = fakeredis.FakeRedis()

# Add locations to a geospatial index
# Format: (longitude, latitude, member_name)
locations = [
    (-74.0059, 40.7128, "New York"),      # NYC coordinates
    (-118.2437, 34.0522, "Los Angeles"),  # LA coordinates
    (-87.6298, 41.8781, "Chicago"),       # Chicago coordinates
    (-122.4194, 37.7749, "San Francisco") # SF coordinates
]

added = client.geoadd("cities", locations)
print(f"Added {added} cities to geospatial index")

# Get coordinates for specific cities
positions = client.geopos("cities", "New York", "Los Angeles", "NonExistent")
for i, pos in enumerate(positions):
    city = ["New York", "Los Angeles", "NonExistent"][i]
    if pos:
        lon, lat = pos
        print(f"{city}: {lon:.4f}, {lat:.4f}")
    else:
        print(f"{city}: Not found")

# Calculate distance between cities
distance_km = client.geodist("cities", "New York", "Los Angeles", unit="km")
distance_mi = client.geodist("cities", "New York", "Los Angeles", unit="mi")
print(f"NYC to LA: {distance_km:.2f} km ({distance_mi:.2f} miles)")

import fakeredis

client = fakeredis.FakeRedis()

# Add some landmarks
landmarks = [
    (-0.1276, 51.5074, "London"),         # London, UK
    (2.3522, 48.8566, "Paris"),           # Paris, France  
    (13.4050, 52.5200, "Berlin"),         # Berlin, Germany
    (12.4964, 41.9028, "Rome")            # Rome, Italy
]

client.geoadd("landmarks", landmarks)

# Get geohashes for the landmarks
hashes = client.geohash("landmarks", "London", "Paris", "Berlin", "Rome")
for i, city in enumerate(["London", "Paris", "Berlin", "Rome"]):
    if hashes[i]:
        print(f"{city}: {hashes[i]}")

# Geohashes can be used for:
# 1. Approximate location representation
# 2. Hierarchical spatial indexing
# 3. Location-based sharding
print(f"\nGeohash precision example:")
london_hash = hashes[0]
if london_hash:
    for precision in [1, 3, 5, 7, 9, 11]:
        if precision <= len(london_hash):
            truncated = london_hash[:precision]
            print(f"  Precision {precision}: {truncated}")

import fakeredis

client = fakeredis.FakeRedis()

# Add restaurants in a city area
restaurants = [
    (-73.9857, 40.7484, "Restaurant A"),    # Times Square area
    (-73.9776, 40.7505, "Restaurant B"),    # Near Times Square
    (-74.0059, 40.7128, "Restaurant C"),    # Downtown
    (-73.9712, 40.7831, "Restaurant D"),    # Upper West Side
    (-73.9442, 40.8176, "Restaurant E")     # Harlem
]

client.geoadd("restaurants", restaurants)

# Find restaurants within 2km of Times Square
times_square_lon, times_square_lat = -73.9857, 40.7484

nearby_simple = client.georadius("restaurants", times_square_lon, times_square_lat, 2, unit="km")
print("Restaurants within 2km of Times Square:")
for restaurant in nearby_simple:
    print(f"  - {restaurant.decode()}")

# Find restaurants with distances and coordinates
nearby_detailed = client.georadius(
    "restaurants", 
    times_square_lon, times_square_lat, 
    2, 
    unit="km",
    withdist=True,
    withcoord=True,
    sort="ASC"  # Sort by distance
)

print("\nDetailed results (with distances and coordinates):")
for result in nearby_detailed:
    name = result[0].decode()
    distance = result[1]
    coords = result[2]
    print(f"  - {name}: {distance:.3f}km at ({coords[0]:.4f}, {coords[1]:.4f})")

import fakeredis

client = fakeredis.FakeRedis()

# Add coffee shops
coffee_shops = [
    (-73.9857, 40.7484, "Starbucks Times Square"),
    (-73.9776, 40.7505, "Local Cafe A"),
    (-73.9712, 40.7831, "Local Cafe B"),
    (-74.0059, 40.7128, "Downtown Coffee"),
    (-73.9442, 40.8176, "Uptown Roasters")
]

client.geoadd("coffee_shops", coffee_shops)

# Find coffee shops within 1.5km of "Starbucks Times Square"
nearby_coffee = client.georadiusbymember(
    "coffee_shops",
    "Starbucks Times Square",
    1.5,
    unit="km",
    withdist=True,
    count=5,
    sort="ASC"
)

print("Coffee shops within 1.5km of Starbucks Times Square:")
for result in nearby_coffee:
    name = result[0].decode()
    distance = result[1]
    if name != "Starbucks Times Square":  # Exclude the reference point
        print(f"  - {name}: {distance:.3f}km away")

import fakeredis

# Use Redis 6.2+ for advanced geosearch features
client = fakeredis.FakeRedis(version=(6, 2))

# Add hotels in a city
hotels = [
    (-73.9857, 40.7484, "Hotel Plaza"),
    (-73.9776, 40.7505, "Boutique Inn"),
    (-74.0059, 40.7128, "Business Hotel"),
    (-73.9712, 40.7831, "Luxury Resort"),
    (-73.9442, 40.8176, "Budget Lodge"),
    (-73.9900, 40.7300, "Downtown Suites"),
    (-73.9600, 40.7700, "Midtown Hotel")
]

client.geoadd("hotels", hotels)

# Search in circular area from coordinates
circular_search = client.geosearch(
    "hotels",
    longitude=-73.9857,
    latitude=40.7484,
    radius=1,
    unit="km",
    withdist=True,
    withcoord=True,
    sort="ASC",
    count=3
)

print("Hotels within 1km (circular search):")
for result in circular_search:
    name = result[0].decode()
    distance = result[1]
    coords = result[2]
    print(f"  - {name}: {distance:.3f}km at ({coords[0]:.4f}, {coords[1]:.4f})")

# Search in rectangular area from a member
rectangular_search = client.geosearch(
    "hotels",
    member="Hotel Plaza",
    width=2,
    height=1,
    unit="km",
    withdist=True,
    sort="ASC"
)

print("\nHotels within 2km x 1km rectangle from Hotel Plaza:")
for result in rectangular_search:
    name = result[0].decode()
    distance = result[1]
    print(f"  - {name}: {distance:.3f}km away")

# Store search results in another key
stored_count = client.geosearchstore(
    "nearby_hotels",        # Destination key
    "hotels",               # Source key
    member="Hotel Plaza",
    radius=1.5,
    unit="km",
    storedist=True         # Store distances as scores
)

print(f"\nStored {stored_count} hotels in 'nearby_hotels' with distances as scores")

# Retrieve stored results (they're stored as a sorted set)
stored_results = client.zrange("nearby_hotels", 0, -1, withscores=True)
print("Stored nearby hotels with distances:")
for member, distance in stored_results:
    print(f"  - {member.decode()}: {distance:.3f}km")

import fakeredis

client = fakeredis.FakeRedis()

# Add initial locations
locations = [
    (-73.9857, 40.7484, "Store_1"),
    (-73.9776, 40.7505, "Store_2"),
    (-74.0059, 40.7128, "Store_3")
]

added = client.geoadd("stores", locations)
print(f"Initially added {added} stores")

# Add new locations (won't overwrite existing)
new_locations = [
    (-73.9857, 40.7484, "Store_1"),     # Duplicate - won't be added
    (-73.9712, 40.7831, "Store_4"),     # New location
    (-73.9442, 40.8176, "Store_5")      # New location
]

added_new = client.geoadd("stores", new_locations, nx=True)  # Only add if not exists
print(f"Added {added_new} new stores (nx=True)")

# Update existing locations
updated_locations = [
    (-73.9860, 40.7480, "Store_1"),     # Slightly different coordinates
    (-73.9780, 40.7500, "Store_2")      # Slightly different coordinates
]

updated = client.geoadd("stores", updated_locations, xx=True)  # Only update existing
print(f"Updated {updated} existing stores (xx=True)")

# Get count of changes
change_locations = [
    (-73.9850, 40.7490, "Store_1"),     # Change existing
    (-73.9000, 40.8000, "Store_6")      # Add new
]

changes = client.geoadd("stores", change_locations, ch=True)  # Return count of changes
print(f"Total changes made: {changes}")

# Verify final positions
all_positions = client.geopos("stores", "Store_1", "Store_2", "Store_3", "Store_4", "Store_5", "Store_6")
for i, pos in enumerate(all_positions):
    store_name = f"Store_{i+1}"
    if pos:
        lon, lat = pos
        print(f"{store_name}: ({lon:.4f}, {lat:.4f})")
    else:
        print(f"{store_name}: Not found")

import fakeredis
import time
import math
from dataclasses import dataclass
from typing import List, Tuple, Optional

@dataclass
class Location:
    id: str
    name: str
    longitude: float
    latitude: float
    category: str
    rating: float = 0.0
    
@dataclass 
class SearchResult:
    location: Location
    distance_km: float
    
class LocationService:
    def __init__(self, client: fakeredis.FakeRedis):
        self.client = client
        
    def add_location(self, location: Location) -> bool:
        """Add a location to the geospatial index"""
        # Store location in geospatial index
        geo_result = self.client.geoadd(
            f"locations:{location.category}",
            [(location.longitude, location.latitude, location.id)]
        )
        
        # Store location metadata
        self.client.hset(f"location:{location.id}", mapping={
            "name": location.name,
            "category": location.category,
            "longitude": str(location.longitude),
            "latitude": str(location.latitude),
            "rating": str(location.rating)
        })
        
        return geo_result > 0
    
    def find_nearby(
        self,
        longitude: float,
        latitude: float,
        radius_km: float,
        category: Optional[str] = None,
        limit: Optional[int] = None
    ) -> List[SearchResult]:
        """Find locations within radius of coordinates"""
        
        categories = [category] if category else self._get_all_categories()
        all_results = []
        
        for cat in categories:
            geo_key = f"locations:{cat}"
            
            # Search for locations in this category
            results = self.client.georadius(
                geo_key,
                longitude,
                latitude,
                radius_km,
                unit="km",
                withdist=True,
                sort="ASC",
                count=limit
            )
            
            # Convert to SearchResult objects
            for result in results:
                location_id = result[0].decode()
                distance = result[1]
                
                # Get location metadata
                location_data = self.client.hgetall(f"location:{location_id}")
                if location_data:
                    location = Location(
                        id=location_id,
                        name=location_data[b'name'].decode(),
                        longitude=float(location_data[b'longitude'].decode()),
                        latitude=float(location_data[b'latitude'].decode()),
                        category=location_data[b'category'].decode(),
                        rating=float(location_data[b'rating'].decode())
                    )
                    
                    all_results.append(SearchResult(location, distance))
        
        # Sort by distance and apply limit
        all_results.sort(key=lambda x: x.distance_km)
        return all_results[:limit] if limit else all_results
    
    def find_nearby_location(
        self,
        reference_location_id: str,
        radius_km: float,
        category: Optional[str] = None,
        limit: Optional[int] = None
    ) -> List[SearchResult]:
        """Find locations near another location"""
        
        # Get reference location data
        ref_data = self.client.hgetall(f"location:{reference_location_id}")
        if not ref_data:
            return []
        
        ref_category = ref_data[b'category'].decode()
        
        # Search in the reference location's category
        results = self.client.georadiusbymember(
            f"locations:{ref_category}",
            reference_location_id,
            radius_km,
            unit="km",
            withdist=True,
            sort="ASC",
            count=limit + 1 if limit else None  # +1 to account for reference location
        )
        
        search_results = []
        for result in results:
            location_id = result[0].decode()
            distance = result[1]
            
            # Skip the reference location itself
            if location_id == reference_location_id:
                continue
                
            # Get location metadata
            location_data = self.client.hgetall(f"location:{location_id}")
            if location_data:
                location = Location(
                    id=location_id,
                    name=location_data[b'name'].decode(),
                    longitude=float(location_data[b'longitude'].decode()),
                    latitude=float(location_data[b'latitude'].decode()),
                    category=location_data[b'category'].decode(),
                    rating=float(location_data[b'rating'].decode())
                )
                
                search_results.append(SearchResult(location, distance))
        
        return search_results[:limit] if limit else search_results
    
    def get_location_stats(self, category: str) -> dict:
        """Get statistics for locations in a category"""
        geo_key = f"locations:{category}"
        
        # Get all members in the geospatial index
        all_members = self.client.zrange(geo_key, 0, -1)
        
        if not all_members:
            return {"total": 0, "average_rating": 0.0}
        
        total_rating = 0.0
        count = 0
        
        for member in all_members:
            location_data = self.client.hgetall(f"location:{member.decode()}")
            if location_data:
                rating = float(location_data[b'rating'].decode())
                total_rating += rating
                count += 1
        
        return {
            "total": count,
            "average_rating": total_rating / count if count > 0 else 0.0
        }
    
    def _get_all_categories(self) -> List[str]:
        """Get all location categories"""
        keys = self.client.keys("locations:*")
        return [key.decode().split(":")[1] for key in keys]

# Usage example
client = fakeredis.FakeRedis()
location_service = LocationService(client)

# Add sample locations
locations = [
    Location("rest_1", "Pizza Palace", -73.9857, 40.7484, "restaurant", 4.2),
    Location("rest_2", "Burger Joint", -73.9776, 40.7505, "restaurant", 3.8),
    Location("rest_3", "Sushi Bar", -73.9712, 40.7831, "restaurant", 4.5),
    Location("hotel_1", "Grand Hotel", -73.9850, 40.7480, "hotel", 4.0),
    Location("hotel_2", "Budget Inn", -73.9780, 40.7500, "hotel", 3.2),
    Location("shop_1", "Fashion Store", -73.9860, 40.7490, "shopping", 4.1),
    Location("shop_2", "Electronics Hub", -73.9770, 40.7510, "shopping", 3.9)
]

for location in locations:
    location_service.add_location(location)

print("Added all locations to the service")

# Find restaurants within 1km of Times Square
times_square = (-73.9857, 40.7484)
nearby_restaurants = location_service.find_nearby(
    times_square[0], times_square[1],
    radius_km=1.0,
    category="restaurant",
    limit=5
)

print(f"\nRestaurants within 1km of Times Square:")
for result in nearby_restaurants:
    loc = result.location
    print(f"  - {loc.name}: {result.distance_km:.3f}km (Rating: {loc.rating})")

# Find locations near Pizza Palace
nearby_pizza_palace = location_service.find_nearby_location(
    "rest_1",  # Pizza Palace
    radius_km=0.5,
    limit=3
)

print(f"\nLocations within 0.5km of Pizza Palace:")
for result in nearby_pizza_palace:
    loc = result.location
    print(f"  - {loc.name} ({loc.category}): {result.distance_km:.3f}km")

# Get statistics for each category
for category in ["restaurant", "hotel", "shopping"]:
    stats = location_service.get_location_stats(category)
    print(f"\n{category.title()} stats:")
    print(f"  Total locations: {stats['total']}")
    print(f"  Average rating: {stats['average_rating']:.2f}")

import fakeredis
import math
from typing import List, Tuple, Dict
from dataclasses import dataclass

@dataclass
class DeliveryZone:
    id: str
    name: str
    center_longitude: float
    center_latitude: float
    radius_km: float
    delivery_fee: float
    min_order: float

class DeliveryService:
    def __init__(self, client: fakeredis.FakeRedis):
        self.client = client
        
    def add_delivery_zone(self, zone: DeliveryZone):
        """Add a delivery zone"""
        # Store zone center in geospatial index
        self.client.geoadd(
            "delivery_zones",
            [(zone.center_longitude, zone.center_latitude, zone.id)]
        )
        
        # Store zone metadata
        self.client.hset(f"zone:{zone.id}", mapping={
            "name": zone.name,
            "center_longitude": str(zone.center_longitude),
            "center_latitude": str(zone.center_latitude),
            "radius_km": str(zone.radius_km),
            "delivery_fee": str(zone.delivery_fee),
            "min_order": str(zone.min_order)
        })
    
    def check_delivery_availability(
        self,
        customer_longitude: float,
        customer_latitude: float
    ) -> List[Dict]:
        """Check which delivery zones serve a customer location"""
        
        # Find all delivery zones within a reasonable search radius (e.g., 50km)
        nearby_zones = self.client.georadius(
            "delivery_zones",
            customer_longitude,
            customer_latitude,
            50,  # Search within 50km
            unit="km",
            withdist=True
        )
        
        available_zones = []
        
        for result in nearby_zones:
            zone_id = result[0].decode()
            distance_to_center = result[1]
            
            # Get zone details
            zone_data = self.client.hgetall(f"zone:{zone_id}")
            if zone_data:
                zone_radius = float(zone_data[b'radius_km'].decode())
                
                # Check if customer is within this zone's delivery radius
                if distance_to_center <= zone_radius:
                    available_zones.append({
                        "zone_id": zone_id,
                        "zone_name": zone_data[b'name'].decode(),
                        "delivery_fee": float(zone_data[b'delivery_fee'].decode()),
                        "min_order": float(zone_data[b'min_order'].decode()),
                        "distance_from_center": distance_to_center
                    })
        
        # Sort by delivery fee (cheapest first)
        available_zones.sort(key=lambda x: x['delivery_fee'])
        return available_zones
    
    def get_optimal_delivery_zone(
        self,
        customer_longitude: float,
        customer_latitude: float,
        order_value: float
    ) -> Dict:
        """Get the best delivery zone for a customer order"""
        
        available_zones = self.check_delivery_availability(
            customer_longitude, customer_latitude
        )
        
        # Filter zones by minimum order requirement
        eligible_zones = [
            zone for zone in available_zones 
            if order_value >= zone['min_order']
        ]
        
        if not eligible_zones:
            return {"available": False, "reason": "No delivery zones available or order below minimum"}
        
        # Return the zone with lowest delivery fee
        best_zone = eligible_zones[0]
        return {
            "available": True,
            "zone": best_zone,
            "total_delivery_fee": best_zone['delivery_fee']
        }
    
    def update_zone_radius(self, zone_id: str, new_radius_km: float):
        """Update delivery zone radius"""
        return self.client.hset(f"zone:{zone_id}", "radius_km", str(new_radius_km))
    
    def get_zone_coverage_stats(self) -> List[Dict]:
        """Get statistics for all delivery zones"""
        # Get all zones
        all_zones = self.client.zrange("delivery_zones", 0, -1)
        
        stats = []
        for zone_member in all_zones:
            zone_id = zone_member.decode()
            zone_data = self.client.hgetall(f"zone:{zone_id}")
            
            if zone_data:
                stats.append({
                    "zone_id": zone_id,
                    "name": zone_data[b'name'].decode(),
                    "radius_km": float(zone_data[b'radius_km'].decode()),
                    "delivery_fee": float(zone_data[b'delivery_fee'].decode()),
                    "min_order": float(zone_data[b'min_order'].decode()),
                    "coverage_area_km2": math.pi * (float(zone_data[b'radius_km'].decode()) ** 2)
                })
        
        return stats

# Usage example
client = fakeredis.FakeRedis()
delivery_service = DeliveryService(client)

# Add delivery zones for a food delivery service
zones = [
    DeliveryZone("zone_downtown", "Downtown", -73.9857, 40.7484, 2.0, 3.99, 15.0),
    DeliveryZone("zone_midtown", "Midtown", -73.9776, 40.7505, 1.5, 2.99, 12.0),
    DeliveryZone("zone_uptown", "Uptown", -73.9712, 40.7831, 3.0, 4.99, 20.0),
    DeliveryZone("zone_financial", "Financial District", -74.0059, 40.7128, 1.0, 1.99, 10.0)
]

for zone in zones:
    delivery_service.add_delivery_zone(zone)

print("Delivery zones configured:")
zone_stats = delivery_service.get_zone_coverage_stats()
for stat in zone_stats:
    print(f"  - {stat['name']}: {stat['radius_km']}km radius, ${stat['delivery_fee']} fee, ${stat['min_order']} minimum, {stat['coverage_area_km2']:.1f}km² coverage")

# Test delivery availability for different customer locations
test_locations = [
    {"name": "Customer A", "lon": -73.9850, "lat": 40.7480, "order_value": 25.0},  # Downtown area
    {"name": "Customer B", "lon": -73.9780, "lat": 40.7510, "order_value": 8.0},   # Midtown, low order
    {"name": "Customer C", "lon": -74.0050, "lat": 40.7130, "order_value": 15.0},  # Financial district
    {"name": "Customer D", "lon": -73.9000, "lat": 40.8000, "order_value": 30.0},  # Far location
]

print(f"\nDelivery availability check:")
for customer in test_locations:
    print(f"\n{customer['name']} (Order: ${customer['order_value']}):")
    
    # Check all available zones
    available = delivery_service.check_delivery_availability(customer['lon'], customer['lat'])
    if available:
        print(f"  Available zones:")
        for zone in available:
            print(f"    - {zone['zone_name']}: ${zone['delivery_fee']} fee, ${zone['min_order']} min, {zone['distance_from_center']:.2f}km from center")
    
    # Get optimal zone
    optimal = delivery_service.get_optimal_delivery_zone(customer['lon'], customer['lat'], customer['order_value'])
    if optimal['available']:
        zone = optimal['zone']
        print(f"  ✅ Best option: {zone['zone_name']} (${optimal['total_delivery_fee']} delivery)")
    else:
        print(f"  ❌ {optimal['reason']}")