Stellen Sie sich eine riesige Open-Pit-Mine vor: Dutzende autonomous Fahrzeuge navigieren gleichzeitig durch Staub, Schlaglöcher und wechselnde Lichtverhältnisse.传统调度方式依赖人工经验,不仅效率低下,还存在严重的安全隐患. In diesem Tutorial zeige ich Ihnen, wie wir mit HolySheep AI eine vollständige无人车调度系统 aufgebaut haben – von der Echtzeit-Oberflächenerkennung mit Gemini bis zur automatisierten Dokumentation mit Claude.
Warum dieser Artikel? Nach 3 Jahren Entwicklungsarbeit in der Bergbau-Automatisierung habe ich hunderte von Stunden mit fehlgeschlagenen API-Integrationen verbracht. Dieser Leitfaden fasst alles zusammen, was ich mir gewünscht hätte, als ich angefangen habe.
Das Problem: Warum klassische Bergbau-Software scheitert
Traditionelle SCADA-Systeme in Minen sind für statische Umgebungen konzipiert. Wenn jedoch 30+ autonome Fahrzeuge gleichzeitig operieren, entstehen folgende Herausforderungen:
- Reaktionszeit: Menschliche Dispatcher brauchen 15-30 Sekunden für Entscheidungen
- Oberflächenqualität: Schlaglöcher und Nässe werden zu spät erkannt
- Dokumentation: Schichtberichte gehen verloren oder sind inkonsistent
- Kosten: Proprietäre Bergbau-Software kostet $50.000+ pro Jahr
Unsere Lösung: Eine Dreifach-Architektur
Wir haben ein System entwickelt, das drei KI-Komponenten nahtlos integriert:
- Gemini 2.5 Flash: Echtzeit-Bildanalyse der Fahrbahnoberfläche
- Claude Sonnet 4.5: Intelligente Dispatch-Entscheidungen und Protokollierung
- MCP Server: Sichere Middleware zwischen Fahrzeugen und Cloud
Grundlagen: API-Zugang einrichten (Schritt für Schritt)
Bevor wir Code schreiben, brauchen Sie Zugang zur HolySheep AI API. Der Prozess dauert weniger als 5 Minuten:
- Registrieren Sie sich auf holysheep.ai/register
- Erhalten Sie 100 kostenlose Credits (ausreichend für 40.000 Bildanalysen)
- Navigieren Sie zu "API Keys" und erstellen Sie einen neuen Schlüssel
- Kopieren Sie den Key – er beginnt mit
hs_
Wichtig: Die Preise bei HolySheep sind unschlagbar. Während GPT-4.1 bei anderen Anbietern $8 pro Million Tokens kostet, zahlen Sie bei HolySheep umgerechnet nur ca. $1,20 – eine 85%+ Ersparnis bei vergleichbarer Qualität.
Komponente 1: Gemini路面识别 (Straßenoberflächenerkennung)
Der erste Baustein unseres Systems ist die Echtzeit-Analyse von Kamerabildern. Gemini 2.5 Flash eignet sich hervorragend dafür, weil er:
- Mehrere Objekte gleichzeitig erkennt
- Bei schlechten Lichtverhältnissen robust arbeitet
- Mit nur $2,50/Million Tokens extrem günstig ist
- Latenzzeiten unter 50ms erreicht
Vollständiger Python-Code für die Straßenerkennung
#!/usr/bin/env python3
"""
HolySheep AI - Gemini路面识别 für autonome Bergbaufahrzeuge
Kompatibel mit Python 3.8+ und Raspberry Pi 4
"""
import base64
import json
import requests
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
============================================
KONFIGURATION
============================================
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class SurfaceCondition(Enum):
"""Zustands-Kategorien für die Fahrbahnoberfläche"""
EXCELLENT = "excellent"
GOOD = "good"
WARNING = "warning"
DANGER = "danger"
UNKNOWN = "unknown"
@dataclass
class RoadAnalysis:
"""Analyseergebnis der Fahrbahnoberfläche"""
condition: SurfaceCondition
confidence: float
defects: List[str]
recommended_speed: int
alternatives: List[str]
class MiningRoadAnalyzer:
"""
Analysiert Kamerabilder von Bergbaufahrzeugen in Echtzeit
Erkennt: Schlaglöcher, Risse, Wasseransammlungen, Steine
"""
SYSTEM_PROMPT = """Du bist ein erfahrener Bergbau-Ingenieur mit 20 Jahren Erfahrung in
der Bewertung von Minenstraßen. Analysiere das Bild und gib zurück:
1. ZUSTAND: excellent/good/warning/danger
2. KONFIDENZ: Zahl zwischen 0.0 und 1.0
3. DEFEKTE: Liste der gefundenen Probleme (auf Chinesisch und Deutsch)
4. EMPFOHLENE_GESCHWINDIGKEIT: Zahl in km/h
5. ALTERNATIV_ROUTEN: Liste möglicher Ausweichrouten
Antworte NUR im JSON-Format, keine Erklärungen."""
def __init__(self, api_key: str):
self.api_key = api_key
self.endpoint = f"{HOLYSHEEP_BASE_URL}/chat/completions"
def analyze_road_surface(
self,
image_path: str,
vehicle_id: str = "UNKNOWN"
) -> RoadAnalysis:
"""
Analysiert ein Bild der Fahrbahnoberfläche
Args:
image_path: Pfad zum Kamerabild
vehicle_id: Fahrzeug-Kennung
Returns:
RoadAnalysis mit Zustandsbewertung
"""
# Bild in Base64 kodieren
with open(image_path, "rb") as img_file:
image_base64 = base64.b64encode(img_file.read()).decode("utf-8")
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gemini-2.5-flash",
"messages": [
{
"role": "system",
"content": self.SYSTEM_PROMPT
},
{
"role": "user",
"content": [
{
"type": "text",
"text": f"Fahrzeug-ID: {vehicle_id}\nAnalysiere die aktuelle Fahrbahnoberfläche:"
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
}
],
"temperature": 0.1, # Niedrig für konsistente Analyse
"max_tokens": 500
}
try:
response = requests.post(
self.endpoint,
headers=headers,
json=payload,
timeout=10
)
response.raise_for_status()
result = response.json()
analysis_text = result["choices"][0]["message"]["content"]
return self._parse_analysis(analysis_text)
except requests.exceptions.Timeout:
return RoadAnalysis(
condition=SurfaceCondition.UNKNOWN,
confidence=0.0,
defects=["Timeout bei API-Anfrage"],
recommended_speed=10,
alternatives=[]
)
except Exception as e:
print(f"Fehler bei Analyse: {e}")
return RoadAnalysis(
condition=SurfaceCondition.UNKNOWN,
confidence=0.0,
defects=[str(e)],
recommended_speed=0,
alternatives=[]
)
def _parse_analysis(self, raw_response: str) -> RoadAnalysis:
"""Parst die JSON-Antwort von Gemini"""
try:
# Versuche, JSON aus der Antwort zu extrahieren
json_str = raw_response.strip()
if "```json" in json_str:
json_str = json_str.split("``json")[1].split("``")[0]
elif "```" in json_str:
json_str = json_str.split("``")[1].split("``")[0]
data = json.loads(json_str)
condition_map = {
"excellent": SurfaceCondition.EXCELLENT,
"good": SurfaceCondition.GOOD,
"warning": SurfaceCondition.WARNING,
"danger": SurfaceCondition.DANGER
}
return RoadAnalysis(
condition=condition_map.get(
data.get("ZUSTAND", "unknown").lower(),
SurfaceCondition.UNKNOWN
),
confidence=float(data.get("KONFIDENZ", 0.0)),
defects=data.get("DEFEKTE", []),
recommended_speed=int(data.get("EMPFOHLENE_GESCHWINDIGKEIT", 30)),
alternatives=data.get("ALTERNATIV_ROUTEN", [])
)
except json.JSONDecodeError:
# Fallback: einfache Textanalyse
return RoadAnalysis(
condition=SurfaceCondition.UNKNOWN,
confidence=0.5,
defects=["Parse-Fehler", raw_response[:100]],
recommended_speed=20,
alternatives=[]
)
============================================
BEISPIEL-NUTZUNG
============================================
if __name__ == "__main__":
analyzer = MiningRoadAnalyzer(HOLYSHEEP_API_KEY)
# Simulierte Analyse (in Produktion: echtes Kamerabild)
result = analyzer.analyze_road_surface(
image_path="/tmp/road_cam_001.jpg",
vehicle_id="AUV-042"
)
print(f"🚗 Fahrzeug: AUV-042")
print(f"📊 Zustand: {result.condition.value}")
print(f"🎯 Konfidenz: {result.confidence:.1%}")
print(f"⚠️ Defekte: {', '.join(result.defects)}")
print(f"🚫 Empfohlene Geschwindigkeit: {result.recommended_speed} km/h")
if result.alternatives:
print(f"🛤️ Alternativrouten: {result.alternatives}")
Wie der Code funktioniert
Screenshot-Hinweis: Öffnen Sie die HolySheep AI Dashboard unter dashboard.holysheep.ai und navigieren Sie zu "Usage Analytics", um Ihre API-Aufrufe in Echtzeit zu verfolgen.
Der Code ist in drei Hauptbereiche unterteilt:
- Konfiguration: API-Key und Base-URL werden zentral definiert
- MiningRoadAnalyzer-Klasse: Encapsulated alle API-Logik
- Beispiel-Nutzung: Demonstriert den typischen Workflow
Komponente 2: Claude调度纪要 (Dispatch-Protokollierung)
Der zweite kritische Baustein ist die automatische Generierung von Schichtprotokollen. Wir nutzen Claude Sonnet 4.5, weil:
- Die Sprachqualität für technische Dokumentation hervorragend ist
- Kontext-Längen bis zu 200K Tokens verarbeitet werden
- Structurierte JSON-Ausgaben zuverlässig generiert werden
- Der Preis von $15/Million Tokens durch HolySheep auf ca. $2,25 sinkt
#!/usr/bin/env python3
"""
HolySheep AI - Claude Dispatch-Protokoll für Minen-Schichtberichte
Automatische Generierung von Schichtprotokollen basierend auf Fahrzeugdaten
"""
import json
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
============================================
KONFIGURATION
============================================
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
@dataclass
class VehicleEvent:
"""Ein Ereignis im Tagesablauf eines Fahrzeugs"""
timestamp: str
vehicle_id: str
event_type: str # LOAD, DUMP, WARNING, MAINTENANCE, etc.
location: str
details: str
duration_minutes: Optional[int] = None
@dataclass
class ShiftReport:
"""Vollständiger Schichtbericht"""
shift_id: str
date: str
shift_type: str # EARLY, LATE, NIGHT
supervisor: str
total_vehicles: int
total_loads: int
total_tonnage: float
incidents: List[Dict]
road_conditions: Dict[str, str]
recommendations: List[str]
class DispatchProtocolGenerator:
"""
Generiert automatisch professionelle Schichtberichte
Basierend auf Fahrzeugdaten und KI-Analyse
"""
SYSTEM_PROMPT = """Du bist ein erfahrener Minen-Dispatcher mit 15 Jahren Erfahrung.
Deine Aufgabe ist es, aus rohen Fahrzeugdaten einen professionellen Schichtbericht zu erstellen.
REGELN:
- Verwende professionelle Bergbau-Terminologie
- Hebe Sicherheitsvorfälle besonders hervor
- Liste alle Fahrzeugbewegungen chronologisch auf
- Berechne KPIs: Auslastung, Effizienz, Stillstandszeiten
- Formuliere konkrete Verbesserungsvorschläge
Antworte NUR mit gültigem JSON im folgenden Format:
{
"ZUSAMMENFASSUNG": "2-3 Sätze zur Gesamtsituation",
"SCHICHT_ID": "2026-05-27-FRÜH",
"DATUM": "27.05.2026",
"SCHICHT": "Frühschicht 06:00-14:00",
"PERSONAL": {
"Dispatcher": "Name",
"Supervisor": "Name",
"Fahrer_aktiv": 12
},
"FAHRZEUGE": {
"AUV-001": {"loads": 15, "tonnage": 450, "status": "aktiv"},
"AUV-002": {"loads": 12, "tonnage": 360, "status": "wartung"}
},
"EREIGNISSE": [
{
"zeit": "08:30",
"typ": "WARUNG",
"fahrzeug": "AUV-042",
"beschreibung": "Schlagloch auf Route B erkannt",
"aktion": "Tempolimit auf 20km/h reduziert"
}
],
"SICHERHEIT": {
"vorfaelle": 2,
"beinahe_unfälle": 0,
"sicherheitsstopps": 1
},
"KPIS": {
"gesamttonnage": 810,
"durchschnittliche_auslastung": 87.5,
"effizienz_bewertung": "gut"
},
"STRASSENBEDINGUNGEN": {
"Route_A": "Gut - keine Beeinträchtigungen",
"Route_B": "Warnung - Schlagloch bei km 2.3",
"Route_C": "Gesperrt - Wartungsarbeiten"
},
"EMPFEHLUNGEN": [
"Route B nach Schichtende reparieren",
"AUV-002 Ölwechsel vor nächster Schicht"
]
}"""
def __init__(self, api_key: str):
self.api_key = api_key
self.endpoint = f"{HOLYSHEEP_BASE_URL}/chat/completions"
def generate_shift_report(
self,
shift_date: datetime,
shift_type: str,
vehicle_events: List[VehicleEvent],
supervisor: str = "System"
) -> ShiftReport:
"""
Generiert einen vollständigen Schichtbericht
Args:
shift_date: Datum der Schicht
shift_type: EARLY, LATE, oder NIGHT
vehicle_events: Liste aller Fahrzeugereignisse
supervisor: Name des Schichtleiters
Returns:
ShiftReport Objekt
"""
# Formatiere die Ereignisse als Text
events_text = self._format_events(vehicle_events)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
shift_name = {
"EARLY": "Frühschicht 06:00-14:00",
"LATE": "Spätschicht 14:00-22:00",
"NIGHT": "Nachtschicht 22:00-06:00"
}.get(shift_type, shift_type)
user_message = f"""
SCHICHTBERICHT GENERIERUNG
Datum: {shift_date.strftime('%d.%m.%Y')}
Schicht: {shift_name}
Schichtleiter: {supervisor}
FAHRZEUGEREIGNISSE (rohe Daten):
{events_text}
Bitte generiere den vollständigen Schichtbericht.
"""
payload = {
"model": "claude-sonnet-4.5",
"messages": [
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": user_message}
],
"temperature": 0.3,
"max_tokens": 2000
}
try:
response = requests.post(
self.endpoint,
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
content = result["choices"][0]["message"]["content"]
return self._parse_report(content, shift_date, shift_type, supervisor)
except Exception as e:
print(f"Fehler bei Berichtsgenerierung: {e}")
return self._create_fallback_report(shift_date, shift_type)
def _format_events(self, events: List[VehicleEvent]) -> str:
"""Formatiert Ereignisse für den Prompt"""
lines = []
for event in sorted(events, key=lambda x: x.timestamp):
line = f"[{event.timestamp}] {event.vehicle_id} | {event.event_type} | {event.location} | {event.details}"
if event.duration_minutes:
line += f" | Dauer: {event.duration_minutes}min"
lines.append(line)
return "\n".join(lines) if lines else "Keine Ereignisse aufgezeichnet"
def _parse_report(self, content: str, shift_date: datetime, shift_type: str, supervisor: str) -> ShiftReport:
"""Parst die Claude-Antwort zu einem ShiftReport"""
try:
json_str = content.strip()
if "```json" in json_str:
json_str = json_str.split("``json")[1].split("``")[0]
data = json.loads(json_str)
return ShiftReport(
shift_id=data.get("SCHICHT_ID", f"{shift_date.date()}-{shift_type}"),
date=data.get("DATUM", shift_date.strftime("%d.%m.%Y")),
shift_type=data.get("SCHICHT", shift_type),
supervisor=supervisor,
total_vehicles=len(data.get("FAHRZEUGE", {})),
total_loads=sum(v.get("loads", 0) for v in data.get("FAHRZEUGE", {}).values()),
total_tonnage=data.get("KPIS", {}).get("gesamttonnage", 0),
incidents=data.get("EREIGNISSE", []),
road_conditions=data.get("STRASSENBEDINGUNGEN", {}),
recommendations=data.get("EMPFEHLUNGEN", [])
)
except Exception as e:
print(f"Parse-Fehler: {e}")
return self._create_fallback_report(shift_date, shift_type)
def _create_fallback_report(self, shift_date: datetime, shift_type: str) -> ShiftReport:
"""Erstellt einen minimalen Fallback-Bericht"""
return ShiftReport(
shift_id=f"ERROR-{shift_date.date()}",
date=shift_date.strftime("%d.%m.%Y"),
shift_type=shift_type,
supervisor="SYSTEM",
total_vehicles=0,
total_loads=0,
total_tonnage=0,
incidents=[],
road_conditions={},
recommendations=["Bericht konnte nicht generiert werden"]
)
============================================
BEISPIEL-NUTZUNG
============================================
if __name__ == "__main__":
generator = DispatchProtocolGenerator(HOLYSHEEP_API_KEY)
# Simulierte Fahrzeugereignisse
sample_events = [
VehicleEvent(
timestamp="2026-05-27T06:15:00",
vehicle_id="AUV-001",
event_type="LOAD",
location="Zone-A-Lader",
details="120 Tonnen Erz geladen",
duration_minutes=8
),
VehicleEvent(
timestamp="2026-05-27T06:45:00",
vehicle_id="AUV-001",
event_type="DUMP",
location=" crusher-station",
details="Entladen an Brecheranlage",
duration_minutes=5
),
VehicleEvent(
timestamp="2026-05-27T08:30:00",
vehicle_id="AUV-042",
event_type="WARNING",
location="Route-B-km-2.3",
details="Schlagloch erkannt - Geschwindigkeit reduziert",
duration_minutes=None
),
VehicleEvent(
timestamp="2026-05-27T10:00:00",
vehicle_id="AUV-002",
event_type="MAINTENANCE",
location="Werkstatt",
details="Geplanter Ölwechsel",
duration_minutes=45
),
]
# Generiere Bericht
report = generator.generate_shift_report(
shift_date=datetime(2026, 5, 27),
shift_type="EARLY",
vehicle_events=sample_events,
supervisor="张伟 (Zhang Wei)"
)
print("=" * 60)
print("📋 SCHICHTBERICHT - GENERIERT")
print("=" * 60)
print(f"Schicht-ID: {report.shift_id}")
print(f"Datum: {report.date}")
print(f"Typ: {report.shift_type}")
print(f"Schichtleiter: {report.supervisor}")
print(f"Gesamt-Tonnage: {report.total_tonnage} t")
print(f"Ereignisse: {len(report.incidents)}")
print(f"Empfehlungen: {len(report.recommendations)}")
print("=" * 60)
Komponente 3: MCP Server工程落地 (MCP Server Engineering)
Der Model Context Protocol (MCP) Server ist das Herzstück unserer Architektur. Er fungiert als sichere Middleware zwischen den Fahrzeugen, der Cloud-Infrastruktur und den KI-Modellen.
Architektur-Übersicht
Screenshot-Hinweis: In der HolySheep AI Console finden Sie unter "MCP Servers" eine visuelle Übersicht der Server-Status. Rote Punkte zeigen Verbindungsprobleme, grüne Punkte bestätigen aktive Verbindungen.
#!/usr/bin/env python3
"""
HolySheep AI - MCP Server für Bergbau-Dispatch-System
Verbindet Fahrzeug-Telemetrie mit KI-Modellen
Stabil, fehlertolerant, production-ready
"""
import asyncio
import json
import logging
import ssl
import uuid
from datetime import datetime
from typing import Dict, Any, Optional, Callable
from dataclasses import dataclass, asdict
from enum import Enum
import hashlib
Optionale Abhängigkeiten
try:
import websockets
from websockets.server import WebSocketServerProtocol
WEBSOCKETS_AVAILABLE = True
except ImportError:
WEBSOCKETS_AVAILABLE = False
try:
import redis.asyncio as redis
REDIS_AVAILABLE = True
except ImportError:
REDIS_AVAILABLE = False
============================================
KONFIGURATION
============================================
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger("MiningDispatch.MCP")
class MessageType(Enum):
"""Unterstützte Nachrichtentypen"""
VEHICLE_TELEMETRY = "vehicle_telemetry"
ROAD_ANALYSIS_REQUEST = "road_analysis_request"
ROAD_ANALYSIS_RESULT = "road_analysis_result"
DISPATCH_COMMAND = "dispatch_command"
HEARTBEAT = "heartbeat"
ERROR = "error"
@dataclass
class MCPMessage:
"""Standard-MCP-Nachrichtenformat"""
id: str
type: str
timestamp: str
source: str
destination: Optional[str]
payload: Dict[str, Any]
def to_json(self) -> str:
return json.dumps({
"id": self.id,
"type": self.type,
"timestamp": self.timestamp,
"source": self.source,
"destination": self.destination,
"payload": self.payload
})
@classmethod
def from_json(cls, data: str) -> 'MCPMessage':
parsed = json.loads(data)
return cls(
id=parsed["id"],
type=parsed["type"],
timestamp=parsed["timestamp"],
source=parsed["source"],
destination=parsed.get("destination"),
payload=parsed["payload"]
)
class HolySheepAIClient:
"""Client für HolySheep AI API mit Retry-Logik"""
def __init__(self, api_key: str, base_url: str):
self.api_key = api_key
self.base_url = base_url
self.endpoint = f"{base_url}/chat/completions"
self._session_count = 0
self._total_tokens = 0
async def analyze_image(
self,
image_base64: str,
vehicle_id: str,
prompt: str = "Analysiere das Bild für Bergbau-Straßenbedingungen"
) -> Dict[str, Any]:
"""Analysiert ein Bild mit Gemini via HolySheep API"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gemini-2.5-flash",
"messages": [
{"role": "user", "content": [
{"type": "text", "text": f"{prompt}\n\nFahrzeug: {vehicle_id}"},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}}
]}
],
"temperature": 0.1,
"max_tokens": 300
}
async with asyncio.Semaphore(10): # Max 10 gleichzeitige Anfragen
for attempt in range(3):
try:
async with asyncio.timeout(15):
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(
self.endpoint,
headers=headers,
json=payload
) as response:
if response.status == 200:
result = await response.json()
self._session_count += 1
tokens = result.get("usage", {}).get("total_tokens", 0)
self._total_tokens += tokens
return result["choices"][0]["message"]["content"]
elif response.status == 429:
wait_time = 2 ** attempt
logger.warning(f"Rate limit erreicht, warte {wait_time}s")
await asyncio.sleep(wait_time)
else:
raise Exception(f"API-Fehler: {response.status}")
except asyncio.TimeoutError:
logger.warning(f"Timeout bei Versuch {attempt + 1}")
if attempt == 2:
raise
except Exception as e:
logger.error(f"Fehler bei Anfrage: {e}")
if attempt == 2:
raise
return {"error": "Max retries exceeded"}
async def generate_text(
self,
system_prompt: str,
user_prompt: str,
model: str = "claude-sonnet-4.5"
) -> str:
"""Generiert Text mit Claude via HolySheep API"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.3,
"max_tokens": 1000
}
try:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(
self.endpoint,
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
result = await response.json()
return result["choices"][0]["message"]["content"]
else:
error_text = await response.text()
raise Exception(f"API-Fehler {response.status}: {error_text}")
except Exception as e:
logger.error(f"Textgenerierung fehlgeschlagen: {e}")
return f"FEHLER: {str(e)}"
class MiningDispatchMCPServer:
"""
MCP Server für Mining Dispatch System
Verbindet Fahrzeuge, KI-Modelle und Dispatcher
"""
def __init__(
self,
host: str = "0.0.0.0",
port: int = 8765,
api_key: str = HOLYSHEEP_API_KEY
):
self.host = host
self.port = port
self.ai_client = HolySheepAIClient(api_key, HOLYSHEEP_BASE_URL)
# Verbindungsverwaltung
self.vehicles: Dict[str, WebSocketServerProtocol] = {}
self.dispatchers: Dict[str, WebSocketServerProtocol] = {}
# Statistiken
self.stats = {
"messages_processed": 0,
"images_analyzed": 0,
"reports_generated": 0,
"errors": 0,
"start_time": datetime.now().isoformat()
}
# Lock für Thread-Safety
self._lock = asyncio.Lock()
logger.info(f"MCP Server initialisiert auf {host}:{port}")
async def handle_vehicle_message(
self,
websocket: WebSocketServerProtocol,
message: MCPMessage
):
"""Behandelt Nachrichten von Fahrzeugen"""
async with self._lock:
if message.type == MessageType.VEHICLE_TELEMETRY.value:
await self._handle_telemetry(websocket, message)
elif message.type == MessageType.ROAD_ANALYSIS_REQUEST.value:
await self._handle_road_analysis(websocket, message)
elif message.type == MessageType.HEARTBEAT.value:
await self._handle_heartbeat(websocket, message)
async def _handle_telemetry(
self,
websocket: WebSocketServerProtocol,
message: MCPMessage
):
"""Verarbeitet Fahrzeug-Telemetrie"""
vehicle_id = message.source
payload = message.payload
logger.info(f"Telemetrie von {vehicle_id}: "
f"Position={payload.get('position')}, "
f"Geschwindigkeit={payload.get('speed')}km/h")
# Weiterleitung an Dispatcher
await self._broadcast_to_dispatchers(message)
self.stats["messages_processed"] += 1
async def _handle_road_analysis(
self,
websocket: WebSocketServerProtocol,
message: MCPMessage
):
"""Analysiert Straßenzustand mit Gemini"""
vehicle_id = message.source
image_data = message.payload.get("image_base64")
if not image_data:
await self._send_error(websocket, message.id, "Keine Bilddaten")
return
logger.info(f"Starte Straßenanalyse für {vehicle_id}")
try:
result = await self.ai_client.analyze_image(
image_base64=image_data,
vehicle_id=vehicle_id,
prompt="Analysiere die Straßenoberfläche für Bergbaufahrzeuge. "
"Erkenne: Schlaglöcher, Risse, Wasser, Steine. "
"Antworte mit: ZUSTAND (excellent/good/warning/danger), "
"DEFEKTE (Liste), EMPFOHLENE_GESCHWINDIGKEIT (km/h)"
)
response = MCPMessage(
id=str(uuid.uuid4()),
type=MessageType.ROAD_ANALYSIS_RESULT.value,
timestamp=datetime.now().isoformat(),
source="MCP_SERVER",
destination=vehicle_id,
payload={
"original_request_id": message.id,
"result": result,
"processing_time_ms": 150 # Beispiel
}
)
await websocket.send(response.to_json())
self.stats["images_analyzed"] += 1
except Exception as e:
logger.error(f"Analysen-Fehler: {e}")
await self._send_error(websocket, message.id, str(e))
self.stats["errors"] += 1
async def _handle_heartbeat(
self,
websocket: WebSocketServerProtocol,
message: MCPMessage
Verwandte Ressourcen
Verwandte Artikel