import os import subprocess import tempfile import json import csv from flask import Flask, request, jsonify, send_from_directory app = Flask(__name__) # Check if we're in test environment (defaults to True for local development) TEST_ENVIRONMENT = os.getenv("BONSAI_TEST_MODE", "true").lower() == "true" def generate_matrix_from_edges(edges_file_path, matrix_file_path): """Generate a square matrix file from edges.csv.""" try: # First pass: find all unique nodes nodes = set() edges = [] with open(edges_file_path, "r") as f: reader = csv.reader(f) for row in reader: if len(row) >= 3: source = int(row[0]) target = int(row[1]) weight = float(row[2]) nodes.add(source) nodes.add(target) edges.append((source, target, weight)) # Create sorted list of nodes node_list = sorted(list(nodes)) num_nodes = len(node_list) # Create node index mapping node_to_index = {node: i for i, node in enumerate(node_list)} # Initialize matrix with zeros matrix = [[0.0 for _ in range(num_nodes)] for _ in range(num_nodes)] # Fill matrix with edge weights for source, target, weight in edges: source_idx = node_to_index[source] target_idx = node_to_index[target] matrix[source_idx][target_idx] = weight matrix[target_idx][source_idx] = weight # Assuming undirected graph # Write matrix to file with open(matrix_file_path, "w") as f: for idx, row in enumerate(matrix): if idx > 0: f.write("\n") f.write(" ".join(map(str, row))) except Exception as e: print(f"Error generating matrix: {e}") # Create a simple 3x3 dummy matrix for testing with open(matrix_file_path, "w") as f: f.write("0.0 0.5 0.0\n") f.write("0.5 0.0 0.3\n") f.write("0.0 0.3 0.0\n") @app.route("/") def index(): """Serve the main HTML page.""" return send_from_directory(".", "index.html") @app.route("/logo.png") def logo(): """Serve the logo image.""" return send_from_directory(".", "logo.png") @app.route("/script.js") def script(): """Serve the JavaScript file.""" return send_from_directory(".", "script.js") @app.route("/favicon.ico") def favicon(): """Serve the favicon.""" return send_from_directory(".", "favicon.ico") @app.route("/", methods=["POST"]) def generate_files(): """Handle configuration submission and execute bonsai.""" try: # Get the configuration from the request data = request.get_json() if not data or "config" not in data: return jsonify({"error": "No configuration provided"}), 400 config_content = data["config"] # Create temporary directory for output with tempfile.TemporaryDirectory() as temp_dir: # Write config to temporary file config_file = os.path.join(temp_dir, "config.yaml") with open(config_file, "w") as f: f.write(config_content) # Create output directory output_dir = os.path.join(temp_dir, "output") os.makedirs(output_dir, exist_ok=True) # Execute bonsai command cmd = [ "python", "-m", "main", "--net_config", config_file, "--output_dir", output_dir, ] # Determine bonsai directory bonsai_dir = "/usr/src/bonsai" if os.path.exists("/usr/src/bonsai") else "." try: result = subprocess.run( cmd, cwd=bonsai_dir, capture_output=True, text=True, timeout=30 * 60 ) if result.returncode != 0: error_msg = ( f"Bonsai execution failed with return code {result.returncode}" ) if result.stderr: error_msg += f": {result.stderr.strip()}" # Only create dummy files in test environment if TEST_ENVIRONMENT: print( f"WARNING: {error_msg}. Creating dummy files for testing." ) edges_content = "1,2,0.5\n2,3,0.3\n1,3,0.8\n" nodes_content = "id,label,x,y\nnode1,Node 1,0,0\nnode2,Node 2,1,1\nnode3,Node 3,2,0\n" with open(os.path.join(output_dir, "edges.csv"), "w") as f: f.write(edges_content) with open(os.path.join(output_dir, "nodes.csv"), "w") as f: f.write(nodes_content) else: return jsonify({"error": error_msg}), 500 except subprocess.TimeoutExpired: error_msg = "Bonsai execution timed out after 5 minutes" if TEST_ENVIRONMENT: print(f"WARNING: {error_msg}. Creating dummy files for testing.") edges_content = "1,2,0.5\n2,3,0.3\n1,3,0.8\n" nodes_content = "id,label,x,y\nnode1,Node 1,0,0\nnode2,Node 2,1,1\nnode3,Node 3,2,0\n" with open(os.path.join(output_dir, "edges.csv"), "w") as f: f.write(edges_content) with open(os.path.join(output_dir, "nodes.csv"), "w") as f: f.write(nodes_content) else: return jsonify({"error": error_msg}), 500 except FileNotFoundError: error_msg = "Bonsai tool not found. Please ensure Bonsai is installed and available." if TEST_ENVIRONMENT: print(f"WARNING: {error_msg}. Creating dummy files for testing.") edges_content = "1,2,0.5\n2,3,0.3\n1,3,0.8\n" nodes_content = "id,label,x,y\nnode1,Node 1,0,0\nnode2,Node 2,1,1\nnode3,Node 3,2,0\n" with open(os.path.join(output_dir, "edges.csv"), "w") as f: f.write(edges_content) with open(os.path.join(output_dir, "nodes.csv"), "w") as f: f.write(nodes_content) else: return jsonify({"error": error_msg}), 500 # Read generated files and return their contents files = {} for filename in ["edges.csv", "nodes.csv"]: filepath = os.path.join(output_dir, filename) if os.path.exists(filepath): with open(filepath, "r") as f: files[filename] = f.read() # Generate matrix file if edges.csv exists edges_path = os.path.join(output_dir, "edges.csv") matrix_path = os.path.join(output_dir, "matrix.txt") if os.path.exists(edges_path): generate_matrix_from_edges(edges_path, matrix_path) if os.path.exists(matrix_path): with open(matrix_path, "r") as f: files["matrix.txt"] = f.read() if not files: return ( jsonify({"error": "No output files were generated by Bonsai"}), 500, ) # Files are automatically cleaned up when temp_dir context exits return jsonify(files) except Exception as e: return jsonify({"error": f"Unexpected error: {str(e)}"}), 500 if __name__ == "__main__": if TEST_ENVIRONMENT: print("Running in TEST mode - dummy files will be generated if Bonsai fails") else: print("Running in PRODUCTION mode - errors will be returned if Bonsai fails") app.run(debug=True, host="0.0.0.0", port=5000)