QGISMCP

#!/usr/bin/env python3 """ QGIS MCP Client - Simple client to connect to the QGIS MCP server """ import socket import json import argparse import sys class QgisMCPClient: def __init__(self, host='localhost', port=9876): self.host = host self.port = port self.socket = None def connect(self): """Connect to the QGIS MCP server""" try: self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.connect((self.host, self.port)) return True except Exception as e: print(f"Error connecting to server: {str(e)}") return False def disconnect(self): """Disconnect from the server""" if self.socket: self.socket.close() self.socket = None def send_command(self, command_type, params=None): """Send a command to the server and get the response""" if not self.socket: print("Not connected to server") return None # Create command command = { "type": command_type, "params": params or {} } try: # Send the command self.socket.sendall(json.dumps(command).encode('utf-8')) # Receive the response response_data = b'' while True: chunk = self.socket.recv(4096) if not chunk: break response_data += chunk # Try to decode as JSON to see if it's complete try: json.loads(response_data.decode('utf-8')) break # Valid JSON, we have the full message except json.JSONDecodeError: continue # Keep receiving # Parse and return the response return json.loads(response_data.decode('utf-8')) except Exception as e: print(f"Error sending command: {str(e)}") return None def ping(self): """Simple ping command to check server connectivity""" return self.send_command("ping") def get_qgis_info(self): """Get QGIS information""" return self.send_command("get_qgis_info") def get_project_info(self): """Get current project information""" return self.send_command("get_project_info") def execute_code(self, code): """Execute arbitrary PyQGIS code""" return self.send_command("execute_code", {"code": code}) def add_vector_layer(self, path, name=None, provider="ogr"): """Add a vector layer to the project""" params = { "path": path, "provider": provider } if name: params["name"] = name return self.send_command("add_vector_layer", params) def add_raster_layer(self, path, name=None, provider="gdal"): """Add a raster layer to the project""" params = { "path": path, "provider": provider } if name: params["name"] = name return self.send_command("add_raster_layer", params) def get_layers(self): """Get all layers in the project""" return self.send_command("get_layers") def remove_layer(self, layer_id): """Remove a layer from the project""" return self.send_command("remove_layer", {"layer_id": layer_id}) def zoom_to_layer(self, layer_id): """Zoom to a layer's extent""" return self.send_command("zoom_to_layer", {"layer_id": layer_id}) def get_layer_features(self, layer_id, limit=10): """Get features from a vector layer""" return self.send_command("get_layer_features", {"layer_id": layer_id, "limit": limit}) def execute_processing(self, algorithm, parameters): """Execute a processing algorithm""" return self.send_command("execute_processing", { "algorithm": algorithm, "parameters": parameters }) def save_project(self, path=None): """Save the current project""" params = {} if path: params["path"] = path return self.send_command("save_project", params) def load_project(self, path): """Load a project""" return self.send_command("load_project", {"path": path}) def render_map(self, path, width=800, height=600): """Render the current map view to an image""" return self.send_command("render_map", { "path": path, "width": width, "height": height }) def print_json(data): """Imprime datos JSON formateados""" print(json.dumps(data, indent=2)) class QgisMCPClient: def process_with_ollama(self, qgis_data, system_prompt=None): """使用Ollama处理QGIS数据 Args: qgis_data: 要处理的QGIS数据 system_prompt: 可选的系统提示 Returns: Ollama的处理结果 """ params = {"qgis_data": qgis_data} if system_prompt: params["system_prompt"] = system_prompt return self.send_command("process_with_ollama", params) def execute_with_ai(self, algorithm, parameters): """使用AI辅助执行处理算法 Args: algorithm: 要执行的算法名称 parameters: 算法参数 Returns: 执行结果和AI分析 """ return self.send_command("execute_with_ai", { "algorithm": algorithm, "parameters": parameters }) def main(): # Conectar al servidor QGIS MCP client = QgisMCPClient(host='localhost', port=9876) if not client.connect(): print("No se pudo conectar al servidor QGIS MCP") return try: # Verificar conexión con ping print("Verificando conexión...") response = client.ping() if response and response.get("status") == "success": print("Conexión exitosa") else: print("Error de conexión") return # Obtener información de QGIS print("\nInformación de QGIS:") qgis_info = client.get_qgis_info() print_json(qgis_info) # Load project print("\nLoad project") load_project = client.load_project("C:/Users/jjsan/OneDrive/Consultoria/Finalizados/electoral_maps/thailand_2007/thailand_2007.qgz") print_json(load_project) # Obtener información del proyecto actual print("\nInformación del proyecto:") project_info = client.get_project_info() print_json(project_info) # Zoom to layer print("\nZoom to first layer") first_layer = project_info["result"]["layers"][0]["id"] zoom_layer = client.zoom_to_layer(first_layer) print_json(zoom_layer) # Render Map to file print("\nRendering image") render_map = client.render_map("C:/Users/jjsan/OneDrive/Consultoria/Finalizados/electoral_maps/thailand_2007/map.png") print_json(render_map) # 添加Ollama处理示例 print("\n使用Ollama处理QGIS数据:") project_info_str = json.dumps(project_info, indent=2) ollama_result = client.process_with_ollama(project_info_str) print_json(ollama_result) # 使用AI辅助执行处理算法示例 print("\n使用AI辅助执行缓冲区分析:") buffer_params = { "INPUT": first_layer, "DISTANCE": 0.01, "SEGMENTS": 5, "END_CAP_STYLE": 0, "JOIN_STYLE": 0, "MITER_LIMIT": 2, "DISSOLVE": False, "OUTPUT": "memory:" } ai_result = client.execute_with_ai("native:buffer", buffer_params) print_json(ai_result) except Exception: print("Error ejecutando comandos") if __name__ == "__main__": main()