Update data.py

data.py

@@ -1,100 +1,192 @@
-import json
+import gradio as gr
+import pandas as pd
+import folium
+from geopy.geocoders import Nominatim
+from geopy.extra.rate_limiter import RateLimiter
+import tempfile
+from typing import Optional, Tuple
+import warnings
 
+# Suppress warnings
+warnings.filterwarnings("ignore")
 
-def extract_leaves(item, path=None, leaves=None):
-    """
-    Extracts the leaves of a nested dictionary or list.
-    """
-    if leaves is None:
-        leaves = []
-    if path is None:
-        path = []
+# Historical Tile Providers
+HISTORICAL_TILES = {
+    "David Rumsey (1790)": {
+        "url": "https://map1.davidrumsey.com/tiles/rumsey/SDSC1790/{z}/{x}/{y}.png",
+        "attr": "David Rumsey Map Collection",
+        "min_year": 1700,
+        "max_year": 1800
+    },
+    "David Rumsey (1860)": {
+        "url": "https://map1.davidrumsey.com/tiles/rumsey/SDSC1860/{z}/{x}/{y}.png",
+        "attr": "David Rumsey Map Collection", 
+        "min_year": 1801,
+        "max_year": 1900
+    },
+    "Stamen (1915)": {
+        "url": "https://stamen-tiles.a.ssl.fastly.net/toner-lite/{z}/{x}/{y}.png",
+        "attr": "Stamen Maps",
+        "min_year": 1901,
+        "max_year": 1920
+    },
+    "OpenHistoricalMap": {
+        "url": "https://tile.openhistoricalmap.org/{z}/{x}/{y}.png",
+        "attr": "OpenHistoricalMap",
+        "min_year": 1700,
+        "max_year": 2023
+    }
+}
 
-    if isinstance(item, dict):
-        for key, value in item.items():
-            extract_leaves(value, path + [key], leaves)
-    elif isinstance(item, list):
-        for value in item:
-            extract_leaves(value, path, leaves)
-    else:
-        if item != '':
-            leaves.append((path, item))
-    return leaves
-
-def split_document(document, window_size, overlap, tokenizer):
-    """
-    Splits a document into chunks of a specified window size with an overlap.
-    """
-    tokens = tokenizer.tokenize(document)
-    print(f"\tLength of document: {len(tokens)} tokens")
-
-    chunks = []
-    if len(tokens) > window_size:
-        for i in range(0, len(tokens), window_size-overlap):
-            print(f"\t{i} to {i + len(tokens[i:i + window_size])}")
-            chunk = tokenizer.convert_tokens_to_string(tokens[i:i + window_size])
-            chunks.append(chunk)
+class Geocoder:
+    def __init__(self):
+        self.geolocator = Nominatim(user_agent="historical_mapper", timeout=10)
+        self.geocode = RateLimiter(self.geolocator.geocode, min_delay_seconds=1)
+        self.cache = {}
+    
+    def get_coords(self, location: str) -> Optional[Tuple[float, float]]:
+        if not location or pd.isna(location):
+            return None
+        if location in self.cache:
+            return self.cache[location]
+        
+        try:
+            result = self.geocode(location)
+            if result:
+                coords = (result.latitude, result.longitude)
+                self.cache[location] = coords
+                return coords
+        except Exception as e:
+            print(f"Geocoding failed for '{location}': {str(e)}")
+        
+        self.cache[location] = None
+        return None
 
-            if i + len(tokens[i:i + window_size]) >= len(tokens):
-                break
-    else:
-        chunks.append(document)
-    print(f"\tSplit into {len(chunks)} chunks")
+def get_tile_layer(year: int):
+    """Select the most appropriate tile layer for the given year"""
+    for name, config in HISTORICAL_TILES.items():
+        if config["min_year"] <= year <= config["max_year"]:
+            return folium.TileLayer(
+                tiles=config["url"],
+                attr=config["attr"],
+                name=name,
+                overlay=False
+            )
+    return folium.TileLayer("OpenStreetMap")
 
-    return chunks
+def create_historical_map(df: pd.DataFrame, location_col: str, year: int = 1900) -> str:
+    geocoder = Geocoder()
+    
+    # Create map with historical base layer
+    base_layer = get_tile_layer(year)
+    m = folium.Map(location=[40, -10], zoom_start=2, control_scale=True)
+    base_layer.add_to(m)
+    
+    # Add all other historical layers as options
+    for name, config in HISTORICAL_TILES.items():
+        if config["url"] != base_layer.tiles:
+            folium.TileLayer(
+                tiles=config["url"],
+                attr=config["attr"],
+                name=f"{name} ({config['min_year']}-{config['max_year']})",
+                overlay=False
+            ).add_to(m)
+    
+    # Add markers with historical styling
+    coords_list = []
+    for loc in df[location_col].dropna().unique():
+        coords = geocoder.get_coords(str(loc))
+        if coords:
+            folium.Marker(
+                location=coords,
+                popup=f"<b>{loc}</b><br>Year: {year}",
+                icon=folium.Icon(
+                    color="red",
+                    icon="info-sign",
+                    prefix="fa"
+                )
+            ).add_to(m)
+            coords_list.append(coords)
+    
+    # Add layer control and fit bounds
+    folium.LayerControl().add_to(m)
+    if coords_list:
+        m.fit_bounds(coords_list)
+    
+    return m._repr_html_()
 
-def handle_broken_output(pred, prev):
-    """
-    Handles broken or empty JSON output by returning the previous prediction.
-    """
+def process_file(file_obj, location_col, year):
     try:
-        if all([(v in ["", []]) for v in json.loads(pred).values()]):
-            # if empty json, return previous
-            pred = prev
-    except:
-        # if broken json, return previous
-        pred = prev
-
-    return pred
-
-def clean_json_text(text):
-    """
-    Cleans JSON text by removing leading/trailing whitespace and escaping special characters.
-    """
-    text = text.strip()
-    text = text.replace("\#", "#").replace("\&", "&")
-    return text
-
-def sync_empty_fields(dict1, dict2):
-    """
-    Synchronize empty fields between two dictionaries.
+        # Read input file
+        df = pd.read_excel(file_obj.name)
+        
+        # Validate column exists
+        if location_col not in df.columns:
+            return None, f"Column '{location_col}' not found", None
+        
+        # Create historical map
+        map_html = create_historical_map(df, location_col, year)
+        
+        # Save processed data
+        with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp:
+            df.to_excel(tmp.name, index=False)
+            processed_path = tmp.name
+        
+        # Generate stats
+        stats = (
+            f"Total locations: {len(df)}\n"
+            f"Unique places: {df[location_col].nunique()}\n"
+            f"Map year: {year}"
+        )
+        
+        return (
+            f"<div style='width:100%; height:70vh'>{map_html}</div>",
+            stats,
+            processed_path
+        )
     
-    Adds empty fields to dict1 based on dict2, or removes them if they don't exist in dict2.
+    except Exception as e:
+        return None, f"Error: {str(e)}", None
+
+# Gradio Interface
+with gr.Blocks(title="Historical Map Explorer", theme=gr.themes.Soft()) as app:
+    gr.Markdown("# Historical Location Mapper")
     
-    Args:
-        dict1 (dict): The dictionary to be modified.
-        dict2 (dict): The reference dictionary with empty fields to be synced.
+    with gr.Row():
+        with gr.Column():
+            file_input = gr.File(
+                label="Upload Excel File",
+                file_types=[".xlsx", ".xls"],
+                type="filepath"
+            )
+            location_col = gr.Textbox(
+                label="Location Column Name",
+                value="locations",
+                placeholder="Enter exact column name with locations"
+            )
+            year = gr.Slider(
+                minimum=1700,
+                maximum=2023,
+                value=1900,
+                step=1,
+                label="Map Year"
+            )
+            map_btn = gr.Button("Generate Historical Map", variant="primary")
         
-    Returns:
-        dict: The modified dict1 with synced empty fields.
-    """
-    if not isinstance(dict1, dict) or not isinstance(dict2, dict):
-        return dict1
+        with gr.Column():
+            map_display = gr.HTML(
+                label="Historical Map",
+                value="<div style='text-align:center;padding:20px;'>"
+                      "Map will appear here after processing</div>"
+            )
+            stats_output = gr.Textbox(label="Statistics")
+            download_output = gr.File(label="Download Processed Data")
 
-    # Traverse dict2 to add or remove empty fields in dict1
-    for key, value in dict2.items():
-        if isinstance(value, dict):  # Handle nested dictionaries
-            dict1[key] = sync_empty_fields(dict1.get(key, {}), value)
-        elif value in (None, "", [], {}):  # Empty field in dict2
-            if key not in dict1:
-                dict1[key] = value  # Add empty field to dict1 if not present
-        else:
-            if key in dict1 and dict1[key] in (None, "", [], {}):
-                del dict1[key]  # Remove empty field from dict1 if not in dict2
-    
-    # Optionally, remove any extra fields in dict1 that are not in dict2
-    keys_to_remove = [key for key in dict1 if key not in dict2]
-    for key in keys_to_remove:
-        del dict1[key]
-    
-    return dict1
+    map_btn.click(
+        process_file,
+        inputs=[file_input, location_col, year],
+        outputs=[map_display, stats_output, download_output]
+    )
+
+if __name__ == "__main__":
+    app.launch()