app.py

import gradio as gr
import geocoder
import pandas as pd
import requests
import urllib.parse
from typing import Dict, Any
from datetime import datetime
import math
from typing import Tuple, List, Optional
import json

class CrimeData:
    def __init__(self, incident_type: str, date_time: datetime, location: Tuple[float, float], 
                 address: str, narrative: str = None):
        self.incident_type = incident_type
        self.date_time = date_time
        self.location = location
        self.address = address
        self.narrative = narrative

def haversine_distance(coord1: Tuple[float, float], coord2: Tuple[float, float]) -> float:
    """Calculate the distance between two coordinates in kilometers."""
    R = 6371.0
    lat1, lon1 = math.radians(coord1[0]), math.radians(coord1[1])
    lat2, lon2 = math.radians(coord2[0]), math.radians(coord2[1])
    
    dlat = lat2 - lat1
    dlon = lon2 - lon1
    
    a = math.sin(dlat/2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2
    c = 2 * math.asin(math.sqrt(a))
    
    return R * c

def geocode_address(address: str, city: str = "Kingston", province: str = "ON") -> Optional[Tuple[float, float]]:
    """Convert address to latitude/longitude coordinates."""
    full_address = f"{address}, {city}, {province}, Canada"
    location = geocoder.osm(full_address, headers={
        'User-Agent': 'CrimeLookupApp/1.0 ([email protected])'
    })
    
    if location.ok:
        return (location.lat, location.lng)
    
    # Fallback: Direct request to Nominatim API
    import requests
    import urllib.parse
    
    url = f"https://nominatim.openstreetmap.org/search"
    params = {
        'q': full_address,
        'format': 'jsonv2',
        'addressdetails': 1,
        'limit': 1
    }
    
    headers = {
        'User-Agent': 'CrimeLookupApp/1.0 ([email protected])'
    }
    
    try:
        response = requests.get(url, params=params, headers=headers)
        response.raise_for_status()
        results = response.json()
        
        if results:
            return (float(results[0]['lat']), float(results[0]['lon']))
    except Exception as e:
        print(f"Geocoding error: {e}")
    
    return None

def load_crime_data() -> List[CrimeData]:
    """Load crime data from the police incidents API."""
    url = "https://ce-portal-service.commandcentral.com/api/v1.0/public/incidents"
    
    # Define the request payload with the Kingston area boundaries
    payload = {
        "limit": 2000,
        "offset": 0,
        "geoJson": {
            "type": "Polygon",
            "coordinates": [[
                [-76.5167293, 44.2255476],
                [-76.5167293, 44.2435476],
                [-76.4987293, 44.2435476],
                [-76.4987293, 44.2255476],
                [-76.5167293, 44.2255476]
            ]]
        },
        "projection": False,
        "propertyMap": {
            "pageSize": "2000",
            "zoomLevel": "15",
            "latitude": "44.2345476",
            "longitude": "-76.5077293",
            "relativeDate": "custom",
            "fromDate": "2024-01-11T14:00:00.000Z",
            "toDate": "2025-01-11T13:00:00.000Z",
            "days": "",
            "startHour": "0",
            "endHour": "24",
            "parentIncidentTypeIds": "149,150,148,8,97,104,165,98,100,179,178,180,101,99,103,163,168,166,12,161,14,16,15,160,121,162,164,167,173,169,170,172,171,151",
            "agencyIds": "407,1358,ottawapolice.ca,kpf.ca"
        }
    }
    
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Origin': 'https://www.cityprotect.com',
        'Referer': 'https://www.cityprotect.com/'
    }
    
    try:
        response = requests.post(url, json=payload, headers=headers)
        response.raise_for_status()
        data = response.json()
        
        crimes = []
        if 'result' in data and 'list' in data['result'] and 'incidents' in data['result']['list']:
            for incident in data['result']['list']['incidents']:
                crimes.append(CrimeData(
                    incident_type=incident['incidentType'],
                    date_time=datetime.fromisoformat(incident['date'].rstrip('Z')),
                    location=(incident['location']['coordinates'][1], incident['location']['coordinates'][0]),
                    address=incident['blockizedAddress'],
                    narrative=incident.get('narrative', '')
                ))
        return crimes
    except Exception as e:
        print(f"Error loading crime data: {e}")
        return []

def filter_crimes(center: Tuple[float, float], radius_km: float, 
                   crimes: List[CrimeData], start_date: datetime, end_date: datetime) -> List[CrimeData]:
    """Filter crimes by distance and date range."""
    return [crime for crime in crimes 
            if (haversine_distance(center, crime.location) <= radius_km and
                start_date <= crime.date_time <= end_date)]

def format_crime_report(crimes: List[CrimeData]) -> pd.DataFrame:
    """Format crimes into a DataFrame for display."""
    data = []
    for crime in crimes:
        data.append({
            'Date': crime.date_time.strftime('%Y-%m-%d'),
            'Time': crime.date_time.strftime('%H:%M'),
            'Type': crime.incident_type,
            'Location': crime.address,
        })
    return pd.DataFrame(data)

def lookup_crimes(address: str, radius: float = 1.0, 
                start_date: str = None, end_date: str = None) -> pd.DataFrame:
    """Main function to lookup crimes near an address within a date range."""
    # Validate and parse dates
    try:
        start = datetime.strptime(start_date, '%Y-%m-%d') if start_date else datetime(2023, 1, 1)
        end = datetime.strptime(end_date, '%Y-%m-%d') if end_date else datetime.now()
    except ValueError:
        return pd.DataFrame({'Error': ['Invalid date format. Please use YYYY-MM-DD']})
        
    # Geocode the address
    coords = geocode_address(address)
    if not coords:
        return pd.DataFrame({'Error': ['Address not found']})
    
    # Load and filter crimes
    all_crimes = load_crime_data()
    filtered_crimes = filter_crimes(coords, radius, all_crimes, start, end)
    
    # Format results
    if not filtered_crimes:
        return pd.DataFrame({'Message': [f'No crimes found in specified radius between {start_date} and {end_date}']})
    
    return format_crime_report(filtered_crimes)

# Create Gradio interface
iface = gr.Interface(
    fn=lookup_crimes,
    inputs=[
        gr.Textbox(label="Address (e.g., '503 Victoria St, Kingston')"),
        gr.Slider(minimum=0.1, maximum=5.0, value=1.0, label="Radius (km)"),
        gr.Textbox(label="Start Date (YYYY-MM-DD)", value="2024-02-01"),
        gr.Textbox(label="End Date (YYYY-MM-DD)", value=datetime.now().strftime('%Y-%m-%d'))
    ],
    outputs=gr.Dataframe(),
    title="Neighborhood Crime Lookup",
    description="Enter an address and date range to see crimes in the area. Add street name and number only - city is assumed to be Kingston, ON.",
    examples=[
        ["503 Victoria St", 1.0, "2024-01-01", "2024-01-10"],
        ["417 Princess St", 0.5, "2024-06-01", "2024-12-31"]
    ]
)

# Launch the app
if __name__ == "__main__":
    iface.launch()