import gradio as gr import json import os import pandas as pd import folium from folium.plugins import MeasureControl, Fullscreen, MarkerCluster from geopy.geocoders import Nominatim from geopy.exc import GeocoderTimedOut, GeocoderServiceError import time import random from typing import List, Tuple, Optional import io import tempfile import warnings import string import spaces from transformers import AutoModelForCausalLM, AutoTokenizer, AutoConfig import torch warnings.filterwarnings("ignore") # Map Tile Providers with reliable sources MAP_TILES = { "GreenMap": { "url": "https://server.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{z}/{y}/{x}", "attr": "Esri" } } # Model configuration - corrected model name MODEL_NAME = "numind/NuExtract-1.5" # Fixed model name according to documentation DEVICE = "cuda" if torch.cuda.is_available() else "cpu" TORCH_DTYPE = torch.bfloat16 if DEVICE == "cuda" else torch.float32 MAX_INPUT_LENGTH = 20000 # For sliding window processing MAX_NEW_TOKENS = 1000 # Global model variables tokenizer = None model = None try: from transformers.models.qwen2.tokenization_qwen2 import Qwen2Tokenizer from transformers.models.qwen2.modeling_qwen2 import Qwen2ForCausalLM print("Qwen2 components successfully imported") except ImportError: print("Could not import Qwen2 components directly") class SafeGeocoder: def __init__(self): user_agent = f"location_mapper_v1_{random.randint(1000, 9999)}" self.geolocator = Nominatim(user_agent=user_agent, timeout=10) self.cache = {} self.last_request = 0 def _respect_rate_limit(self): current_time = time.time() elapsed = current_time - self.last_request if elapsed < 1.0: time.sleep(1.0 - elapsed) self.last_request = current_time def get_coords(self, location: str): if not location or pd.isna(location): return None location = str(location).strip() if location in self.cache: return self.cache[location] try: self._respect_rate_limit() result = self.geolocator.geocode(location) if result: coords = (result.latitude, result.longitude) self.cache[location] = coords return coords self.cache[location] = None return None except Exception as e: print(f"Geocoding error for '{location}': {e}") self.cache[location] = None return None def process_excel(file, places_column): if file is None: return None, "No file uploaded", None try: if hasattr(file, 'name'): df = pd.read_excel(file.name) elif isinstance(file, bytes): df = pd.read_excel(io.BytesIO(file)) else: df = pd.read_excel(file) print(f"Spalten in der Excel-Tabelle: {list(df.columns)}") if places_column not in df.columns: return None, f"Spalte '{places_column}' wurde in der Excel-Datei nicht gefunden. Verfügbare Spalten: {', '.join(df.columns)}", None # Create a new DataFrame to store expanded rows expanded_rows = [] geocoder = SafeGeocoder() coords = [] processed_count = 0 location_cache = {} # Cache to ensure consistent coordinates # Process each row for idx, row in df.iterrows(): if pd.isna(row[places_column]): # Keep rows with no location as-is expanded_rows.append(row.to_dict()) continue location = str(row[places_column]).strip() try: locations = [loc.strip() for loc in location.split(',') if loc.strip()] if not locations: locations = [location] except: locations = [location] # Process each location in the comma-separated list location_rows_added = False for loc in locations: # Use cached coordinates if available if loc in location_cache: point = location_cache[loc] else: point = geocoder.get_coords(loc) if point: location_cache[loc] = point # Cache the result if point: # Create a new row for this location new_row = row.copy() new_row_dict = new_row.to_dict() new_row_dict[places_column] = loc # Replace with just this location new_row_dict['latitude'] = point[0] new_row_dict['longitude'] = point[1] # Add the row to our expanded rows list expanded_rows.append(new_row_dict) coords.append(point) processed_count += 1 location_rows_added = True # If none of the locations could be geocoded, keep the original row if not location_rows_added: expanded_rows.append(row.to_dict()) # Convert the list of dictionaries to a DataFrame expanded_df = pd.DataFrame(expanded_rows) # Create the map m = folium.Map( location=[20, 0], zoom_start=2, control_scale=True ) folium.TileLayer( tiles=MAP_TILES["GreenMap"]["url"], attr=MAP_TILES["GreenMap"]["attr"], name="GreenMap", overlay=False, control=False ).add_to(m) Fullscreen().add_to(m) MeasureControl(position='topright', primary_length_unit='kilometers').add_to(m) # Add markers directly here marker_cluster = MarkerCluster(name="Locations").add_to(m) # Track which coordinates we've already placed on the map seen_coords = {} for idx, row in expanded_df.iterrows(): if 'latitude' in row and 'longitude' in row and not pd.isna(row['latitude']) and not pd.isna(row['longitude']): location = row[places_column] if not pd.isna(row[places_column]) else "Unknown" # Round coordinates to reduce small differences rounded_lat = round(row['latitude'], 5) rounded_lng = round(row['longitude'], 5) coord_key = f"{rounded_lat},{rounded_lng}" # Skip if we've already added a marker at this location if coord_key in seen_coords: continue seen_coords[coord_key] = True additional_info = "" for col in expanded_df.columns: if col not in [places_column, 'latitude', 'longitude'] and not pd.isna(row[col]): additional_info += f"
{col}: {row[col]}" popup_content = f"""

{location}

{additional_info}
""" folium.Marker( location=(row['latitude'], row['longitude']), popup=folium.Popup(popup_content, max_width=300), tooltip=location, icon=folium.Icon(color="blue", icon="info-sign") ).add_to(marker_cluster) if coords: m.fit_bounds(coords) custom_css = """ """ m.get_root().header.add_child(folium.Element(custom_css)) # Save the expanded DataFrame to Excel with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: processed_path = tmp.name expanded_df.to_excel(processed_path, index=False) total_locations = df[places_column].count() success_rate = (processed_count / total_locations * 100) if total_locations > 0 else 0 stats = f"Gefunden: {processed_count} von {total_locations} Orten ({success_rate:.1f}%)" return m._repr_html_(), stats, processed_path except Exception as e: import traceback trace = traceback.format_exc() print(f"Error processing file: {e}\n{trace}") return None, f"Fehler bei der Verarbeitung der Datei: {str(e)}", None # Corrected model loading function based on official usage example @spaces.GPU def extract_info(template, text): global tokenizer, model # Load tokenizer if not loaded yet if tokenizer is None: print("Tokenizer not loaded yet, loading now...") try: try: from modelscope import AutoTokenizer as MSAutoTokenizer tokenizer = MSAutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=True) print("Loaded tokenizer using modelscope AutoTokenizer") except: # Fall back to regular tokenizer tokenizer = AutoTokenizer.from_pretrained( MODEL_NAME, trust_remote_code=True, revision="main" ) print("Loaded tokenizer using standard AutoTokenizer") except Exception as e: trace = traceback.format_exc() print(f"Error loading tokenizer: {e}\n{trace}") return "❌ Fehler beim Laden des Tokenizers", f"{str(e)}" try: # Load model if not loaded yet if model is None: print("Model not loaded yet, loading now...") try: model = AutoModelForCausalLM.from_pretrained( MODEL_NAME, torch_dtype=TORCH_DTYPE, trust_remote_code=True, revision="main", device_map="auto" # Let the model decide CUDA placement ).eval() print(f"✅ Model loaded successfully") except Exception as e: trace = traceback.format_exc() print(f"Error loading model: {e}\n{trace}") return f"❌ Fehler beim Laden des Modells: {str(e)}", "{}" print("Using model for inference...") # Format the template as proper JSON with indentation template_formatted = json.dumps(json.loads(template), indent=4) # Create prompt prompt = f"<|input|>\n### Template:\n{template_formatted}\n### Text:\n{text}\n\n<|output|>" # Tokenize with proper settings inputs = tokenizer( [prompt], return_tensors="pt", truncation=True, padding=True, max_length=MAX_INPUT_LENGTH ).to(model.device) # Use model's device # Generate output with torch.no_grad() for efficiency with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=MAX_NEW_TOKENS, temperature=0.0, do_sample=False ) # Decode the result result_text = tokenizer.decode(outputs[0], skip_special_tokens=True) # Extract the output part if "<|output|>" in result_text: json_text = result_text.split("<|output|>")[1].strip() else: json_text = result_text # Try to parse as JSON try: extracted = json.loads(json_text) return "✅ Erfolgreich extrahiert", json.dumps(extracted, indent=2) except json.JSONDecodeError: return "❌ JSON Parsing Fehler", json_text except Exception as e: import traceback trace = traceback.format_exc() print(f"Error in extract_info: {e}\n{trace}") return f"❌ Fehler: {str(e)}", "{}" @spaces.GPU def create_map(df, location_col): m = folium.Map( location=[20, 0], zoom_start=2, control_scale=True ) folium.TileLayer( tiles=MAP_TILES["GreenMap"]["url"], attr=MAP_TILES["GreenMap"]["attr"], name="GreenMap", overlay=False, control=False ).add_to(m) Fullscreen().add_to(m) MeasureControl(position='topright', primary_length_unit='kilometers').add_to(m) geocoder = SafeGeocoder() coords = [] marker_cluster = MarkerCluster(name="Locations").add_to(m) processed_count = 0 for idx, row in df.iterrows(): if pd.isna(row[location_col]): continue location = str(row[location_col]).strip() additional_info = "" for col in df.columns: if col != location_col and not pd.isna(row[col]): additional_info += f"
{col}: {row[col]}" try: locations = [loc.strip() for loc in location.split(',') if loc.strip()] if not locations: locations = [location] except: locations = [location] for loc in locations: point = geocoder.get_coords(loc) if point: popup_content = f"""

{loc}

{additional_info}
""" folium.Marker( location=point, popup=folium.Popup(popup_content, max_width=300), tooltip=loc, icon=folium.Icon(color="blue", icon="info-sign") ).add_to(marker_cluster) coords.append(point) processed_count += 1 if coords: m.fit_bounds(coords) custom_css = """ """ m.get_root().header.add_child(folium.Element(custom_css)) return m._repr_html_(), processed_count custom_css = """ """ with gr.Blocks(css=custom_css, title="Daten Strukturieren und Analysieren") as demo: gr.HTML("""

Strukturierung und Visualisierung von historischen Daten

In dieser Unterrichtseinheit befassen wir uns mit der Strukturierung unstrukturierter historischer Texte und der Visualisierung von extrahierten Daten auf Karten. Die systematische Strukturierung von Daten wird mit einem für Informationsextrahierung trainiertem Sprachmodell durchgeführt, das auf der Question-Answering-Methode basiert. Diese Methode erlaubt es, Informationen mit Hilfe einer Frage zu extrahieren, wie etwa „Wo fand das Erdbeben statt"? Dies ermöglicht die Extrahierung des Ortes, an dem ein Erdbeben stattfand, auch wenn im Text selbst noch andere Orte genannt werden.

Die Katastrophe in Japan — 3 Millionen Tote. Mtb. London, 4. Sept. (Drahtbericht.) Zu dem Unglück in Japan liegen noch folgende Nachrichten vor: Wie die japanische Gesandtschaft in Peking meldet, sind Unterhandlungen mit China über die sofortige Lieferung von Lebensmitteln ausgenommen worden. Von Peking seien amerikanische, englische und italienische Schiffe mit Lebensmitteln nach Japan abgegangen.
Ort des Erdbebens: Japan
Andere Orte: London, Peking, China

Methodik: Vom unstrukturierten Text zur strukturierten Information

Die grundlegende Herausforderung bei der Arbeit mit historischen Quellen ist, dass relevante Informationen in langen Fließtexten eingebettet sind und manuell mühsam extrahiert werden müssen. Dieser Ansatz automatisiert diesen Prozess.

Wie funktioniert die Informationsextraktion?

  1. Template-Definition: Sie definieren ein JSON-Template mit den Informationstypen, die Sie extrahieren möchten: 
    {"earthquake location": "", "dateline location": ""}
  2. Question-Answering-Methode: Das Sprachmodell interpretiert jedes leere Feld als implizite Frage:
    • "earthquake location": "" → "Wo ist das Erdbeben passiert?"
    • "dateline location": "" → "Von wo wird berichtet?"
  3. Sprachmodell-Verarbeitung: Das NuExtract-1.5 Modell (ein Sequence-to-Sequence Transformer) analysiert den Text vollständig und identifiziert die relevanten Informationen für jedes Template-Feld.
  4. Strukturierte Ausgabe: Das Modell füllt das Template mit den extrahierten Informationen: 
    {"earthquake location": "Japan, Yokohama", "dateline location": "Tokio"}

Technische Funktionsweise des Sprachmodells

Intern erfolgt die Verarbeitung in mehreren Schritten:

  1. Tokenisierung: Der Text wird in bearbeitbare Einheiten zerlegt.
  2. Kontextuelle Analyse: Der Transformer-Mechanismus ermöglicht die Analyse von Beziehungen zwischen allen Textteilen gleichzeitig.
  3. Selektive Aufmerksamkeit: Das Modell fokussiert sich auf Textpassagen, die Antworten auf die impliziten Fragen enthalten könnten.
  4. Generierung: Die erkannten Informationen werden in das vorgegebene Template eingefügt.

Die Kartierungsfunktion

Nach der Extraktion der Ortsangaben ermöglicht unsere Anwendung die automatische Visualisierung dieser Daten auf einer interaktiven Karte:

  1. Geokodierung: Die extrahierten Ortsnamen werden mittels eines geografischen Dienstes in geografische Koordinaten (Längen- und Breitengrade) umgewandelt.
  2. Kartenerstellung: Die Koordinaten werden auf einer interaktiven Karte platziert, wobei jeder Ort durch einen Marker dargestellt wird.
  3. Kontextinformationen: Beim Klick auf einen Marker werden zusätzliche Informationen aus dem Originaltext angezeigt.
  4. Räumliche Analyse: Die Karte ermöglicht die visuelle Analyse der räumlichen Verteilung historischer Ereignisse.

Dieser kombinierte Ansatz aus Textextraktion und geografischer Visualisierung eröffnet neue Möglichkeiten für die räumliche Analyse historischer Quellen und erlaubt es, geografische Muster zu erkennen, die in den reinen Textdaten nicht unmittelbar sichtbar wären.

Diese Methode ermöglicht die effiziente Extraktion und Visualisierung historischer Daten aus unstrukturierten Quellen.

""") with gr.Tabs() as tabs: with gr.TabItem("🔍 Text Extrahierung"): gr.HTML("""

Extrahieren Sie strukturierte Daten aus unstrukturiertem Text

Verwenden Sie das Sprachmodell NuExtract-1.5 um automatisch Informationen zu extrahieren.

""") # Add model loading button and status at the top with gr.Row(): with gr.Column(): template = gr.Textbox( label="JSON Template", value='{"earthquake location": "", "dateline location": "", "source of information": "", "communication form": ""}', lines=5 ) text = gr.Textbox( label="Hier unstrukturierten Text einfügen", value="Die Zahl der Opfer in Japan. Paris, 12. Sept. Der japanische Konsul in Marseille veröffentlicht nachstehendes offizielles Telegramm, das er heute aus Japan erhalten hat: „Die Zahl der Toten beträgt in Tokio laut einer von der Polizei vorgenommenen Zählung mehr als 60000. Die Zahl der Verwundeten beläuft sich auf ungefähr 500000. In Jokohama beträgt die Zahl der Opfer 110 000, was ungefähr ein Viertel der gesamten Bevölkerung dieser Stadt ausmacht. In den Bezirken von Chiba und Kanagama ist die Zahl der Opfer gleichfalls beträchtlich, doch wurde die Zählung noch nicht zu Ende geführt", lines=8 ) extract_btn = gr.Button("Extrahieren Sie Informationen", variant="primary") with gr.Column(): status = gr.Textbox(label="Status") output = gr.Textbox(label="Output", lines=10) excel_download_file = gr.File( label="Excel-Vorlage herunterladen", value="Earthquake_Japan_Analysis.xlsx", # Replace with the actual path to your Excel file visible=True, interactive=False ) extract_btn.click( fn=extract_info, inputs=[template, text], outputs=[status, output] ) with gr.TabItem("📍 Visualisierung von strukturierten Daten"): gr.HTML("""

Visualisieren Sie Daten auf Karten

Laden Sie eine Excel-Tabelle hoch und erstelle eine interaktive Karte.

""") with gr.Row(): with gr.Column(): excel_file = gr.File( label="Upload Excel File", file_types=[".xlsx", ".xls"], elem_classes="file-upload-box" ) places_column = gr.Textbox( label="Name der Tabellenspalte", value="earthquake_locations", placeholder="Füge den Namen der Spalte mit den Orten ein" ) process_btn = gr.Button("Erstellen Sie die Karte", variant="primary") stats_output = gr.Textbox( label="Status", lines=2, elem_classes="stats-box" ) processed_file = gr.File( label="Bearbeitete Daten herunterladen", visible=True, interactive=False ) with gr.Column(): map_output = gr.HTML( label="Interaktive Karte", value="""

Your map will appear here after processing

""", elem_id="map-container" ) def process_and_map(file, column): if file is None: return None, "Hier bitte die Excel-Tabelle hochladen", None try: map_html, stats, processed_path = process_excel(file, column) if map_html and processed_path: responsive_html = f"""
{map_html}
""" return responsive_html, stats, processed_path else: return None, stats, None except Exception as e: import traceback trace = traceback.format_exc() print(f"Error in process_and_map: {e}\n{trace}") return None, f"Error: {str(e)}", None process_btn.click( fn=process_and_map, inputs=[excel_file, places_column], outputs=[map_output, stats_output, processed_file] ) gr.HTML("""

Made with ❤ for historical research

""") if __name__ == "__main__": demo.launch(share=True)