915 lines
36 KiB
Python
915 lines
36 KiB
Python
# -----------------------------------------------------------------------------
|
|
# Description: Flask backend API server for authentication, project management, PDK library access, layout preview, and GDS build endpoints.
|
|
# Inside functions: no_cache_response, login_required_json, wrapper, request_ip, record_action, safe_name, user_layout_root, project_root, cell_file_path, cell_svg_path, cell_routes_path, write_route_points_sidecar, project_gds_path, technology_manifest_path_for_project, current_pdk_root, current_pdk_registry, scoped_pdk_root_for_project, pdk_root_for_request_project, project_meta_path, read_project_meta
|
|
# Developer : Qin Yue @ 2026
|
|
# Organization : OptiHK Limited
|
|
# -----------------------------------------------------------------------------
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import json
|
|
import yaml
|
|
from collections import OrderedDict
|
|
from functools import wraps
|
|
from flask import Flask, jsonify, send_from_directory, request, redirect, url_for, session, render_template, make_response
|
|
from werkzeug.security import check_password_hash
|
|
import database
|
|
from flask import Response
|
|
from gds_builder import build_project_gds
|
|
from layout_preview import create_layout_svg_from_gds
|
|
from pdk_registry import PdkRegistry
|
|
from pdk_access import (
|
|
cleanup_expired_exports,
|
|
create_export_path,
|
|
pdk_root_for_session,
|
|
prefer_full_gds_for_session,
|
|
)
|
|
from routed_layout_preview import create_routed_layout_svg, layout_has_links
|
|
from technology_manifest import TechnologyManifestError, read_technology_manifest
|
|
|
|
# --- Path Configurations ---
|
|
# Centralize all filesystem roots used by Flask routes, previews, project
|
|
# storage, PDK discovery, icon serving, and temporary export downloads.
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
FRONTEND_DIR = os.path.join(BASE_DIR, '..', 'frontend')
|
|
|
|
REPO_ROOT = os.path.abspath(os.path.join(BASE_DIR, '..', '..'))
|
|
PDK_PUBLIC_ROOT = os.path.abspath(os.environ.get(
|
|
'MXPIC_PDK_PUBLIC_ROOT',
|
|
os.path.join(REPO_ROOT, 'opt_pdk_public', 'foundries')
|
|
))
|
|
EDA_PDK_ROOT = os.path.abspath(os.path.join(BASE_DIR, '..', 'mxpic', 'PDKs'))
|
|
YML_PATH = os.path.join(BASE_DIR, '..', 'mxpic', 'PDKs', 'Silterra', 'directories.yaml')
|
|
# Component/category icons are served from backend/icons for the library panel.
|
|
ICONS_DIR = os.path.join(BASE_DIR, 'icons')
|
|
|
|
# Saved project YAML, generated previews, and temporary exports live under the
|
|
# database folder so each user can have isolated project storage.
|
|
DATABASE_ROOT = os.path.abspath(os.path.join(BASE_DIR, '..', 'database'))
|
|
EXPORT_ROOT = os.path.abspath(os.path.join(DATABASE_ROOT, '_exports'))
|
|
|
|
|
|
# Flask serves the HTML/CSS/JS frontend and exposes JSON APIs for persistence,
|
|
# PDK lookup, preview generation, and GDS export.
|
|
app = Flask(__name__, template_folder=FRONTEND_DIR, static_folder=FRONTEND_DIR)
|
|
app.secret_key = os.environ.get('MXPIC_SECRET_KEY', 'change_me_for_intranet_deployment')
|
|
app.config.update(
|
|
SESSION_COOKIE_HTTPONLY=True,
|
|
SESSION_COOKIE_SAMESITE='Lax',
|
|
SESSION_COOKIE_SECURE=os.environ.get('MXPIC_COOKIE_SECURE', '0').lower() in {'1', 'true', 'yes'},
|
|
)
|
|
app.json.sort_keys = False
|
|
|
|
# Initialize the lightweight local database when the server process starts.
|
|
database.init_db()
|
|
|
|
|
|
def no_cache_response(response):
|
|
"""Prevent stale editor assets while canvas features are being revised."""
|
|
response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0'
|
|
response.headers['Pragma'] = 'no-cache'
|
|
response.headers['Expires'] = '0'
|
|
return response
|
|
|
|
|
|
def login_required_json(view_func):
|
|
"""Wrap API handlers so unauthenticated requests receive JSON errors."""
|
|
@wraps(view_func)
|
|
def wrapper(*args, **kwargs):
|
|
"""Run the wrapped route only when a user session is present."""
|
|
if 'user_id' not in session:
|
|
return jsonify({"error": "Authentication required"}), 401
|
|
return view_func(*args, **kwargs)
|
|
return wrapper
|
|
|
|
|
|
def request_ip():
|
|
"""Extract the best available client IP address for audit logging."""
|
|
forwarded_for = request.headers.get('X-Forwarded-For', '')
|
|
if forwarded_for:
|
|
return forwarded_for.split(',')[0].strip()
|
|
return request.remote_addr
|
|
|
|
|
|
def record_action(action, project=None, cell=None, detail=None):
|
|
"""Write a user action into the audit log using current session context."""
|
|
if 'user_id' not in session:
|
|
return
|
|
if isinstance(detail, (dict, list)):
|
|
detail = json.dumps(detail, ensure_ascii=False)
|
|
elif detail is not None:
|
|
detail = str(detail)
|
|
database.add_user_log(
|
|
session.get('user_id'),
|
|
session.get('username', 'unknown'),
|
|
action,
|
|
project=safe_name(project, '') if project else None,
|
|
cell=safe_name(cell, '') if cell else None,
|
|
detail=detail,
|
|
ip_address=request_ip()
|
|
)
|
|
|
|
|
|
def safe_name(value, fallback):
|
|
"""Keep user/project/cell names filesystem-friendly without changing display names."""
|
|
value = (value or '').strip()
|
|
if not value:
|
|
value = fallback
|
|
value = re.sub(r'[^A-Za-z0-9_.-]+', '_', value)
|
|
value = value.strip('._')
|
|
return value or fallback
|
|
|
|
|
|
def user_layout_root():
|
|
"""Return the current user layout directory under the database root."""
|
|
username = safe_name(session.get('username'), 'anonymous')
|
|
return os.path.join(DATABASE_ROOT, username, 'layout')
|
|
|
|
|
|
def project_root(project_name):
|
|
"""Return the filesystem directory for a named project."""
|
|
return os.path.join(user_layout_root(), safe_name(project_name, 'project_1'))
|
|
|
|
|
|
def cell_file_path(project_name, cell_name):
|
|
"""Return the YAML file path for a project cell."""
|
|
return os.path.join(project_root(project_name), f"{safe_name(cell_name, 'canvas_1')}.yml")
|
|
|
|
|
|
def cell_svg_path(project_name, cell_name):
|
|
"""Return the SVG preview path for a project cell."""
|
|
return os.path.join(project_root(project_name), f"{safe_name(cell_name, 'canvas_1')}.svg")
|
|
|
|
|
|
def cell_routes_path(project_name, cell_name):
|
|
"""Return the route sidecar JSON path for a project cell."""
|
|
return os.path.join(project_root(project_name), f"{safe_name(cell_name, 'canvas_1')}.routes.yml")
|
|
|
|
|
|
def write_route_points_sidecar(yaml_content, output_path):
|
|
"""Extract route points from layout YAML and save them beside the cell."""
|
|
layout = yaml.safe_load(yaml_content) or {}
|
|
routes = {}
|
|
# The sidecar preserves manually edited route control points separately from
|
|
# the main YAML file for tooling that wants route-only metadata.
|
|
for bundle_name, bundle in (layout.get("bundles") or {}).items():
|
|
saved_links = []
|
|
for link in bundle.get("links") or []:
|
|
points = link.get("points") or []
|
|
if not points:
|
|
continue
|
|
saved_links.append({
|
|
"from": link.get("from"),
|
|
"to": link.get("to"),
|
|
"xsection": link.get("xsection"),
|
|
"width": link.get("width"),
|
|
"radius": link.get("radius"),
|
|
"points": points,
|
|
})
|
|
if saved_links:
|
|
routes[bundle_name] = {"links": saved_links}
|
|
if routes:
|
|
with open(output_path, 'w', encoding='utf-8') as file:
|
|
yaml.safe_dump({"routes": routes}, file, sort_keys=False)
|
|
elif os.path.exists(output_path):
|
|
os.remove(output_path)
|
|
|
|
|
|
def project_gds_path(project_name):
|
|
"""Return the generated GDS path for a project."""
|
|
return os.path.join(project_root(project_name), f"{safe_name(project_name, 'project_1')}.gds")
|
|
|
|
|
|
def technology_manifest_path_for_project(project_name):
|
|
"""Return the stored technology manifest path for a project."""
|
|
technology_id = read_project_meta(project_name).get("technology") or ""
|
|
if "/" not in technology_id:
|
|
return None
|
|
foundry, technology = technology_id.split("/", 1)
|
|
path = os.path.abspath(os.path.join(EDA_PDK_ROOT, safe_name(foundry, ''), safe_name(technology, ''), "technology.yml"))
|
|
# Keep stored project metadata from escaping the local EDA PDK root.
|
|
if path.startswith(EDA_PDK_ROOT + os.sep) and os.path.exists(path):
|
|
return path
|
|
return None
|
|
|
|
|
|
def current_pdk_root():
|
|
"""Resolve the active PDK root for the current request session."""
|
|
return pdk_root_for_session(session, REPO_ROOT)
|
|
|
|
|
|
def current_pdk_registry():
|
|
"""Create a PDK registry configured for the current session scope."""
|
|
return PdkRegistry(current_pdk_root(), prefer_full_gds=prefer_full_gds_for_session(session))
|
|
|
|
|
|
def scoped_pdk_root_for_project(project_name):
|
|
"""Resolve the correct PDK root for an existing project and session."""
|
|
base_root = current_pdk_root()
|
|
project = safe_name(project_name, '')
|
|
if not project:
|
|
return base_root
|
|
|
|
# When a project has a saved foundry/technology, library browsing is scoped
|
|
# to that technology folder; otherwise it falls back to the user's full PDK
|
|
# access root.
|
|
technology_id = read_project_meta(project).get("technology") or ""
|
|
if "/" not in technology_id:
|
|
return base_root
|
|
|
|
foundry, technology = technology_id.split("/", 1)
|
|
scoped_root = os.path.abspath(os.path.join(base_root, foundry, technology))
|
|
if scoped_root == base_root or not scoped_root.startswith(base_root + os.sep):
|
|
return base_root
|
|
return scoped_root if os.path.isdir(scoped_root) else base_root
|
|
|
|
|
|
def pdk_root_for_request_project():
|
|
"""Resolve the PDK root requested by a project-aware API call."""
|
|
project = request.args.get('project')
|
|
return scoped_pdk_root_for_project(project) if project else current_pdk_root()
|
|
|
|
|
|
def project_meta_path(project_name):
|
|
"""Return the metadata JSON path for a project."""
|
|
return os.path.join(project_root(project_name), ".project.json")
|
|
|
|
|
|
def read_project_meta(project_name):
|
|
"""Read project metadata such as foundry and technology selections."""
|
|
path = project_meta_path(project_name)
|
|
if not os.path.exists(path):
|
|
return {}
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
|
|
|
|
def write_project_meta(project_name, meta):
|
|
"""Persist project metadata to disk."""
|
|
os.makedirs(project_root(project_name), exist_ok=True)
|
|
with open(project_meta_path(project_name), 'w', encoding='utf-8') as f:
|
|
json.dump(meta, f, indent=2)
|
|
|
|
|
|
def ensure_project_path(project_name):
|
|
"""Create and return the directory for a project."""
|
|
layout_root = os.path.abspath(user_layout_root())
|
|
target = os.path.abspath(project_root(project_name))
|
|
# All project paths must remain under the authenticated user's layout root.
|
|
if target != layout_root and not target.startswith(layout_root + os.sep):
|
|
raise ValueError("Invalid project path")
|
|
return target
|
|
|
|
# --- PDK Library Scanning Helpers ---
|
|
# These helpers turn the active PDK folder structure into the nested component
|
|
# library tree shown in the canvas sidebar.
|
|
|
|
def findComps(baseDir, path_root=None):
|
|
"""Scan component folders, return map of paths -> component info."""
|
|
compMap = {}
|
|
refDir = baseDir
|
|
path_root = os.path.abspath(path_root or baseDir)
|
|
# A folder containing a YAML file is treated as a component leaf; scanning
|
|
# stops below that leaf so nested assets do not pollute the library tree.
|
|
for root, dirs, files in os.walk(baseDir):
|
|
ymlFiles = [f for f in files if f.endswith('.yml')]
|
|
if ymlFiles:
|
|
parentDir = os.path.dirname(root)
|
|
relPath = os.path.relpath(parentDir, refDir)
|
|
parts = () if relPath == '.' else tuple(relPath.split(os.sep))
|
|
compName = os.path.basename(root)
|
|
|
|
# Extract the category (the mother folder's name)
|
|
category = os.path.basename(parentDir)
|
|
|
|
# Include compName in the key so multiple cells in one category do not overwrite each other.
|
|
compMap[parts + (compName,)] = {
|
|
'folder': compName,
|
|
'yml': ymlFiles[0],
|
|
'category': category, # Save the category to the map
|
|
'path': os.path.relpath(root, path_root).replace(os.sep, '/')
|
|
}
|
|
dirs.clear()
|
|
return compMap
|
|
|
|
def addCompsToTree(compMap):
|
|
"""Build a completely fresh tree from scratch and insert component nodes."""
|
|
fresh_tree = OrderedDict()
|
|
|
|
# Convert path tuples collected from disk into the nested object structure
|
|
# consumed by the frontend library panel.
|
|
for mapKey, compItem in compMap.items():
|
|
pathSeg = mapKey[:-1]
|
|
compName = compItem['folder']
|
|
curNode = fresh_tree
|
|
|
|
for seg in pathSeg:
|
|
if seg not in curNode:
|
|
curNode[seg] = OrderedDict()
|
|
curNode = curNode[seg]
|
|
|
|
curNode[compName] = OrderedDict({
|
|
"__type__": "component",
|
|
"__name__": compName,
|
|
"__yml__": compItem['yml'],
|
|
"__category__": compItem['category'], # Inject category into the tree
|
|
"__path__": compItem.get('path')
|
|
})
|
|
|
|
return fresh_tree
|
|
|
|
# Component metadata lookup helpers used by library/detail API endpoints.
|
|
|
|
# --- API ROUTES (Library, Components & Icons) ---
|
|
|
|
@app.route('/api/icon/<category>')
|
|
def getIcon(category):
|
|
"""Serve the icon corresponding to the component category."""
|
|
# Prefer exact category artwork, then fall back to a default icon or a
|
|
# transparent 1x1 image to keep the frontend layout stable.
|
|
for ext in ('.png', '.svg', '.jpg'):
|
|
icon_path = os.path.join(ICONS_DIR, f"{category}{ext}")
|
|
if os.path.exists(icon_path):
|
|
return send_from_directory(ICONS_DIR, f"{category}{ext}")
|
|
|
|
fallback = os.path.join(ICONS_DIR, "default.png")
|
|
if os.path.exists(fallback):
|
|
return send_from_directory(ICONS_DIR, "default.png")
|
|
|
|
# return png if not found
|
|
transparent_png = (
|
|
b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01'
|
|
b'\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01'
|
|
b'\x00\x00\x05\x00\x01\r\n\xf4\xc0\x00\x00\x00\x00IEND\xaeB`\x82'
|
|
)
|
|
return Response(transparent_png, mimetype='image/png')
|
|
|
|
# Component metadata helpers sit near the library routes because they share the
|
|
# same role-scoped PDK root resolution.
|
|
|
|
def readCompYaml(compName, comps_root=None):
|
|
"""Load YAML from component folder."""
|
|
search_root = comps_root or current_pdk_root()
|
|
# Component names are resolved by folder basename for compatibility with
|
|
# saved canvas component references.
|
|
for root, dirs, files in os.walk(search_root):
|
|
if os.path.basename(root) == compName:
|
|
dirs.clear()
|
|
ymlFiles = [f for f in files if f.endswith('.yml')]
|
|
if ymlFiles:
|
|
ymlPath = os.path.join(root, ymlFiles[0])
|
|
with open(ymlPath, 'r', encoding='utf-8') as f:
|
|
data = yaml.safe_load(f) or {}
|
|
normalized_root = os.path.abspath(search_root).replace("\\", "/").lower()
|
|
pdk_ports_allowed = "/opt_pdk_public/" in f"{normalized_root}/" or "/opt_pdk_atlas/" in f"{normalized_root}/"
|
|
if pdk_ports_allowed and isinstance(data, dict) and "pins" not in data and isinstance(data.get("ports"), dict):
|
|
data = dict(data)
|
|
data["pins"] = data["ports"]
|
|
return data
|
|
return None
|
|
|
|
|
|
def find_component_dir(component_name, comps_root=None):
|
|
"""Find the directory containing a named component in the PDK library."""
|
|
search_root = comps_root or current_pdk_root()
|
|
for root, dirs, files in os.walk(search_root):
|
|
if os.path.basename(root) == component_name:
|
|
dirs.clear()
|
|
return root, files
|
|
return None, []
|
|
|
|
# --- AUTHENTICATION & PAGE ROUTES ---
|
|
@app.route('/')
|
|
def home():
|
|
"""Route to login page, or bypass to dashboard if already authenticated."""
|
|
if 'user_id' in session:
|
|
return redirect(url_for('dashboard'))
|
|
return render_template('login.html')
|
|
|
|
@app.route('/login', methods=['POST'])
|
|
def login():
|
|
"""Verify credentials against the database."""
|
|
username = request.form.get('username')
|
|
password = request.form.get('password')
|
|
|
|
user = database.get_user(username)
|
|
|
|
# Verify hash from database matches entered password
|
|
if user and check_password_hash(user[2], password):
|
|
session['user_id'] = user[0]
|
|
session['username'] = user[1]
|
|
session['user_group'] = user[3] or 'user'
|
|
record_action('login')
|
|
return redirect(url_for('dashboard'))
|
|
else:
|
|
return render_template('login.html', error="Invalid username or password")
|
|
|
|
@app.route('/dashboard')
|
|
def dashboard():
|
|
"""User project list."""
|
|
if 'user_id' not in session:
|
|
return redirect(url_for('home'))
|
|
|
|
return render_template('dashboard.html', username=session['username'])
|
|
|
|
@app.route('/canvas')
|
|
def canvas():
|
|
"""The main EDA editor."""
|
|
if 'user_id' not in session:
|
|
return redirect(url_for('home'))
|
|
|
|
# Note: Ensure your old index.html is renamed to canvas.html in the frontend folder
|
|
return no_cache_response(make_response(render_template('canvas.html')))
|
|
|
|
|
|
@app.route('/canvas-helpers.js')
|
|
def canvas_helpers():
|
|
"""Serve the shared canvas helper script used by canvas.html."""
|
|
return no_cache_response(send_from_directory(FRONTEND_DIR, 'canvas-helpers.js'))
|
|
|
|
@app.route('/logout')
|
|
def logout():
|
|
"""Clear session and return to login."""
|
|
record_action('logout')
|
|
session.clear()
|
|
return redirect(url_for('home'))
|
|
|
|
|
|
@app.route('/api/health')
|
|
def health_check():
|
|
"""Expose a small deployment health endpoint."""
|
|
return jsonify({"status": "ok", "service": "mxpic_eda"})
|
|
|
|
|
|
@app.route('/api/technologies', methods=['GET'])
|
|
@login_required_json
|
|
def list_technologies():
|
|
"""List technology choices from mxpic/PDKs/<foundry>/<technology>."""
|
|
technologies = []
|
|
pdks_root = EDA_PDK_ROOT
|
|
# Technology choices are built from directory names because each technology
|
|
# folder owns its generated technology.yml manifest.
|
|
if os.path.isdir(pdks_root):
|
|
for foundry in sorted(os.listdir(pdks_root)):
|
|
foundry_path = os.path.join(pdks_root, foundry)
|
|
if not os.path.isdir(foundry_path):
|
|
continue
|
|
for technology in sorted(os.listdir(foundry_path)):
|
|
technology_path = os.path.join(foundry_path, technology)
|
|
if not os.path.isdir(technology_path):
|
|
continue
|
|
technologies.append({
|
|
"foundry": foundry,
|
|
"technology": technology,
|
|
"id": f"{foundry}/{technology}",
|
|
"label": f"{foundry} / {technology}"
|
|
})
|
|
return jsonify({"technologies": technologies})
|
|
|
|
|
|
@app.route('/api/technologies/<foundry>/<technology>/manifest', methods=['GET'])
|
|
@login_required_json
|
|
def get_technology_manifest(foundry, technology):
|
|
"""Return the routing and layer manifest for a selected technology."""
|
|
try:
|
|
manifest = read_technology_manifest(
|
|
EDA_PDK_ROOT,
|
|
safe_name(foundry, ''),
|
|
safe_name(technology, '')
|
|
)
|
|
return jsonify({"manifest": manifest})
|
|
except TechnologyManifestError as e:
|
|
return jsonify({"error": str(e)}), 404
|
|
|
|
|
|
@app.route('/api/profile', methods=['GET', 'PATCH'])
|
|
@login_required_json
|
|
def account_profile():
|
|
"""Return or update profile details for the current user."""
|
|
# Keep the frontend occupation selector constrained to known values that are
|
|
# meaningful for the account page.
|
|
occupations = {'intern', 'senior engineer', 'junior engineer', 'principle engineer'}
|
|
user_id = session.get('user_id')
|
|
|
|
if request.method == 'PATCH':
|
|
data = request.get_json(silent=True) or {}
|
|
occupation = (data.get('occupation') or '').strip().lower()
|
|
if occupation not in occupations:
|
|
return jsonify({"error": "Invalid occupation"}), 400
|
|
database.update_user_occupation(user_id, occupation)
|
|
record_action('profile.update_occupation', detail={"occupation": occupation})
|
|
|
|
profile = database.get_user_profile(user_id)
|
|
if not profile:
|
|
return jsonify({"error": "Profile not found"}), 404
|
|
|
|
return jsonify({
|
|
"id": f"mx-{int(profile[0]):06d}",
|
|
"username": profile[1],
|
|
"created_at": profile[2],
|
|
"credits": profile[3] or 0,
|
|
"occupation": profile[4] or "intern",
|
|
"user_group": profile[5] or "user",
|
|
"occupations": sorted(occupations)
|
|
})
|
|
|
|
|
|
@app.route('/api/profile/password', methods=['POST'])
|
|
@login_required_json
|
|
def change_password():
|
|
"""Validate and persist a password change for the current user."""
|
|
data = request.get_json(silent=True) or {}
|
|
current_password = data.get('current_password') or ''
|
|
new_password = data.get('new_password') or ''
|
|
if len(new_password) < 6:
|
|
return jsonify({"error": "New password must be at least 6 characters"}), 400
|
|
|
|
user = database.get_user_auth_by_id(session.get('user_id'))
|
|
if not user or not check_password_hash(user[2], current_password):
|
|
return jsonify({"error": "Current password is incorrect"}), 400
|
|
|
|
database.update_user_password(user[0], new_password)
|
|
record_action('profile.change_password')
|
|
return jsonify({"message": "Password updated"})
|
|
|
|
|
|
@app.route('/api/logs', methods=['GET', 'POST'])
|
|
@login_required_json
|
|
def user_logs():
|
|
"""Return recent account activity for the current user."""
|
|
if request.method == 'POST':
|
|
data = request.get_json(silent=True) or {}
|
|
action = safe_name(data.get('action'), '')
|
|
if not action:
|
|
return jsonify({"error": "Action is required"}), 400
|
|
record_action(
|
|
action,
|
|
project=data.get('project'),
|
|
cell=data.get('cell'),
|
|
detail=data.get('detail')
|
|
)
|
|
return jsonify({"message": "logged"})
|
|
|
|
try:
|
|
limit = min(500, max(1, int(request.args.get('limit', 200))))
|
|
except ValueError:
|
|
limit = 200
|
|
rows = database.list_user_logs(session.get('user_id'), limit=limit)
|
|
return jsonify({
|
|
"logs": [
|
|
{
|
|
"id": row[0],
|
|
"action": row[1],
|
|
"project": row[2],
|
|
"cell": row[3],
|
|
"detail": row[4],
|
|
"ip_address": row[5],
|
|
"created_at": row[6]
|
|
}
|
|
for row in rows
|
|
]
|
|
})
|
|
|
|
|
|
@app.route('/api/projects', methods=['GET'])
|
|
@login_required_json
|
|
def list_projects():
|
|
"""List projects stored under database/<username>/layout."""
|
|
root = user_layout_root()
|
|
os.makedirs(root, exist_ok=True)
|
|
|
|
projects = []
|
|
# Each project is a folder and each YAML file inside that folder is treated
|
|
# as one saved cell/canvas.
|
|
for name in sorted(os.listdir(root)):
|
|
path = os.path.join(root, name)
|
|
if not os.path.isdir(path):
|
|
continue
|
|
cells = []
|
|
for filename in sorted(os.listdir(path)):
|
|
if not filename.lower().endswith(('.yml', '.yaml')):
|
|
continue
|
|
cell_name = os.path.splitext(filename)[0]
|
|
yml_path = os.path.join(path, filename)
|
|
cells.append({
|
|
"name": cell_name,
|
|
"has_layout": os.path.exists(yml_path)
|
|
})
|
|
meta = read_project_meta(name)
|
|
projects.append({
|
|
"name": name,
|
|
"cells": cells,
|
|
"technology": meta.get("technology")
|
|
})
|
|
|
|
return jsonify({"projects": projects})
|
|
|
|
|
|
@app.route('/api/projects', methods=['POST'])
|
|
@login_required_json
|
|
def create_project():
|
|
"""Create a new project folder and initial cell layout."""
|
|
data = request.get_json(silent=True) or {}
|
|
requested_name = safe_name(data.get('name'), 'project_1')
|
|
technology = data.get('technology') or ''
|
|
root = user_layout_root()
|
|
os.makedirs(root, exist_ok=True)
|
|
|
|
# Preserve the user's requested base name, adding a numeric suffix only when
|
|
# a project folder already exists.
|
|
project_name = requested_name
|
|
counter = 1
|
|
while os.path.exists(os.path.join(root, project_name)):
|
|
counter += 1
|
|
project_name = f"{requested_name}_{counter}"
|
|
|
|
os.makedirs(project_root(project_name), exist_ok=True)
|
|
write_project_meta(project_name, {
|
|
"name": project_name,
|
|
"technology": technology
|
|
})
|
|
record_action('project.create', project=project_name, detail={"technology": technology})
|
|
return jsonify({"name": project_name, "technology": technology}), 201
|
|
|
|
|
|
@app.route('/api/projects/<project_name>', methods=['GET'])
|
|
@login_required_json
|
|
def get_project(project_name):
|
|
"""Load all saved cells for a project."""
|
|
root = project_root(project_name)
|
|
if not os.path.isdir(root):
|
|
return jsonify({"error": "Project not found"}), 404
|
|
|
|
cells = []
|
|
for filename in sorted(os.listdir(root)):
|
|
if not filename.lower().endswith(('.yml', '.yaml')):
|
|
continue
|
|
cell_name = os.path.splitext(filename)[0]
|
|
yml_path = os.path.join(root, filename)
|
|
if not os.path.exists(yml_path):
|
|
continue
|
|
with open(yml_path, 'r', encoding='utf-8') as f:
|
|
cells.append({
|
|
"name": cell_name,
|
|
"content": f.read()
|
|
})
|
|
|
|
return jsonify({
|
|
"name": safe_name(project_name, 'project_1'),
|
|
"cells": cells,
|
|
"technology": read_project_meta(project_name).get("technology")
|
|
})
|
|
|
|
|
|
@app.route('/api/projects/<project_name>', methods=['DELETE'])
|
|
@login_required_json
|
|
def delete_project(project_name):
|
|
"""Delete a user's project folder under database/<username>/layout."""
|
|
root = project_root(project_name)
|
|
layout_root = os.path.abspath(user_layout_root())
|
|
target = os.path.abspath(root)
|
|
|
|
if not target.startswith(layout_root + os.sep):
|
|
return jsonify({"error": "Invalid project path"}), 400
|
|
if not os.path.isdir(target):
|
|
return jsonify({"error": "Project not found"}), 404
|
|
|
|
shutil.rmtree(target)
|
|
record_action('project.delete', project=project_name)
|
|
return jsonify({"message": "deleted", "project": safe_name(project_name, 'project_1')})
|
|
|
|
|
|
@app.route('/api/projects/<project_name>/cells/<cell_name>', methods=['PATCH', 'DELETE'])
|
|
@login_required_json
|
|
def rename_cell(project_name, cell_name):
|
|
"""Rename a project cell and its preview/route sidecar files."""
|
|
if request.method == 'DELETE':
|
|
cell = safe_name(cell_name, 'canvas_1')
|
|
target = os.path.abspath(cell_file_path(project_name, cell))
|
|
project_dir = os.path.abspath(project_root(project_name))
|
|
if not target.startswith(project_dir + os.sep):
|
|
return jsonify({"error": "Invalid cell path"}), 400
|
|
if not os.path.exists(target):
|
|
record_action('canvas.delete_missing', project=project_name, cell=cell)
|
|
return jsonify({"message": "already deleted", "cell": cell})
|
|
os.remove(target)
|
|
record_action('canvas.delete', project=project_name, cell=cell)
|
|
return jsonify({"message": "deleted", "cell": cell})
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
old_cell = safe_name(cell_name, 'canvas_1')
|
|
new_cell = safe_name(data.get('name'), old_cell)
|
|
if old_cell == new_cell:
|
|
return jsonify({"message": "unchanged", "cell": new_cell})
|
|
|
|
old_path = cell_file_path(project_name, old_cell)
|
|
new_path = cell_file_path(project_name, new_cell)
|
|
if os.path.exists(new_path):
|
|
return jsonify({"error": "Cell name already exists"}), 409
|
|
if os.path.exists(old_path):
|
|
os.rename(old_path, new_path)
|
|
|
|
record_action('canvas.rename', project=project_name, cell=new_cell, detail={"old_cell": old_cell})
|
|
return jsonify({"message": "renamed", "old_cell": old_cell, "cell": new_cell})
|
|
|
|
|
|
|
|
|
|
@app.route('/api/save-layout', methods=['POST'])
|
|
@login_required_json
|
|
def save_layout():
|
|
"""Persist a canvas layout YAML document and refresh its preview assets."""
|
|
try:
|
|
data = request.get_json()
|
|
project = safe_name(data.get('project'), 'project_1')
|
|
cell = safe_name(data.get('cell'), 'canvas_1')
|
|
content = data.get('content', '')
|
|
create_preview = bool(data.get('preview', True))
|
|
|
|
save_path = cell_file_path(project, cell)
|
|
os.makedirs(os.path.dirname(save_path), exist_ok=True)
|
|
|
|
with open(save_path, 'w', encoding='utf-8') as f:
|
|
f.write(content)
|
|
write_route_points_sidecar(content, cell_routes_path(project, cell))
|
|
|
|
svg_path = None
|
|
if create_preview:
|
|
svg_path = cell_svg_path(project, cell)
|
|
# Routed layouts need the router backend; placement-only layouts can
|
|
# use the simpler GDS/SVG preview path.
|
|
if layout_has_links(content):
|
|
create_routed_layout_svg(
|
|
content,
|
|
svg_path,
|
|
pdk_root=current_pdk_root(),
|
|
project_dir=project_root(project),
|
|
technology_manifest_path=technology_manifest_path_for_project(project),
|
|
prefer_full_gds=prefer_full_gds_for_session(session),
|
|
)
|
|
else:
|
|
create_layout_svg_from_gds(content, svg_path, pdk_registry=current_pdk_registry(), project_dir=project_root(project))
|
|
|
|
record_action('layout.save', project=project, cell=cell, detail={"bytes": len(content), "svg": svg_path})
|
|
return jsonify({
|
|
"message": "successfully saved",
|
|
"project": project,
|
|
"cell": cell,
|
|
"path": save_path,
|
|
"svg_path": svg_path,
|
|
"svg_url": url_for('get_layout_svg', project_name=project, cell_name=cell) if svg_path else None
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route('/api/projects/<project_name>/cells/<cell_name>/layout.svg')
|
|
@login_required_json
|
|
def get_layout_svg(project_name, cell_name):
|
|
"""Serve a saved SVG layout preview for a project cell."""
|
|
try:
|
|
project_dir = ensure_project_path(project_name)
|
|
svg_path = os.path.abspath(cell_svg_path(project_name, cell_name))
|
|
if not svg_path.startswith(project_dir + os.sep):
|
|
return jsonify({"error": "Invalid SVG path"}), 400
|
|
if not os.path.exists(svg_path):
|
|
return jsonify({"error": "Layout SVG not found"}), 404
|
|
return no_cache_response(send_from_directory(os.path.dirname(svg_path), os.path.basename(svg_path), mimetype='image/svg+xml'))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.route('/api/build-gds', methods=['POST'])
|
|
@login_required_json
|
|
def build_gds():
|
|
"""Build project GDS output and return a downloadable export URL."""
|
|
data = request.get_json(silent=True) or {}
|
|
project = safe_name(data.get('project'), 'project_1')
|
|
try:
|
|
# Expire old exports before creating a new temporary download folder.
|
|
cleanup_expired_exports(EXPORT_ROOT)
|
|
project_dir = ensure_project_path(project)
|
|
if not os.path.isdir(project_dir):
|
|
return jsonify({"error": "Project not found"}), 404
|
|
export_id, filename, output_path = create_export_path(EXPORT_ROOT, safe_name(project, 'project_1'))
|
|
result = build_project_gds(
|
|
project_dir,
|
|
output_path,
|
|
current_pdk_root(),
|
|
technology_manifest_path=technology_manifest_path_for_project(project),
|
|
prefer_full_gds=prefer_full_gds_for_session(session),
|
|
)
|
|
record_action('gds.build', project=project, detail={
|
|
"path": result.output_path,
|
|
"engine": result.engine,
|
|
"warnings": result.warnings
|
|
})
|
|
return jsonify({
|
|
"message": "GDS built",
|
|
"project": project,
|
|
"path": result.output_path,
|
|
"filename": filename,
|
|
"download_url": url_for('download_export', export_id=export_id, filename=filename),
|
|
"engine": result.engine,
|
|
"cells_built": result.cells_built,
|
|
"warnings": result.warnings
|
|
})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route('/api/exports/<export_id>/<filename>')
|
|
@login_required_json
|
|
def download_export(export_id, filename):
|
|
"""Serve a temporary exported GDS file by export id."""
|
|
export_id = safe_name(export_id, '')
|
|
safe_filename = safe_name(filename, 'layout.gds')
|
|
export_dir = os.path.abspath(os.path.join(EXPORT_ROOT, export_id))
|
|
# Export downloads are confined to the temporary export root and removed
|
|
# after the response closes.
|
|
if not export_id or not export_dir.startswith(EXPORT_ROOT + os.sep):
|
|
return jsonify({"error": "Invalid export path"}), 400
|
|
if not safe_filename.lower().endswith('.gds'):
|
|
return jsonify({"error": "Invalid GDS filename"}), 400
|
|
gds_path = os.path.abspath(os.path.join(export_dir, safe_filename))
|
|
if not gds_path.startswith(export_dir + os.sep) or not os.path.exists(gds_path):
|
|
return jsonify({"error": "GDS export not found"}), 404
|
|
response = send_from_directory(export_dir, safe_filename, as_attachment=True)
|
|
response.call_on_close(lambda: shutil.rmtree(export_dir, ignore_errors=True))
|
|
return response
|
|
|
|
|
|
@app.route('/api/projects/<project_name>/gds/<filename>')
|
|
@login_required_json
|
|
def get_project_gds(project_name, filename):
|
|
"""Serve the latest generated GDS file for a project."""
|
|
try:
|
|
project_dir = ensure_project_path(project_name)
|
|
safe_filename = safe_name(filename, f"{safe_name(project_name, 'project_1')}.gds")
|
|
if not safe_filename.lower().endswith('.gds'):
|
|
return jsonify({"error": "Invalid GDS filename"}), 400
|
|
gds_path = os.path.abspath(os.path.join(project_dir, safe_filename))
|
|
if not gds_path.startswith(project_dir + os.sep):
|
|
return jsonify({"error": "Invalid GDS path"}), 400
|
|
if not os.path.exists(gds_path):
|
|
return jsonify({"error": "GDS not found"}), 404
|
|
return send_from_directory(project_dir, safe_filename, as_attachment=True)
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
|
|
# --- API ROUTES (Library & Components) ---
|
|
@app.route('/api/library')
|
|
@login_required_json
|
|
def getLib():
|
|
"""Get library structure."""
|
|
comps_root = pdk_root_for_request_project()
|
|
fresh_tree = {}
|
|
# The library tree is rebuilt on request so project technology changes and
|
|
# role-scoped PDK roots are reflected immediately.
|
|
if os.path.isdir(comps_root):
|
|
compMap = findComps(comps_root, current_pdk_root())
|
|
fresh_tree = addCompsToTree(compMap)
|
|
return jsonify(fresh_tree)
|
|
|
|
|
|
|
|
@app.route('/api/component/<component_name>')
|
|
@login_required_json
|
|
def getComp(component_name):
|
|
"""Return component YAML data."""
|
|
data = readCompYaml(component_name, pdk_root_for_request_project())
|
|
if data is None:
|
|
return jsonify({"error": "Component not found"}), 404
|
|
return jsonify(data)
|
|
|
|
@app.route('/api/component/<component_name>/image')
|
|
@login_required_json
|
|
def getCompImg(component_name):
|
|
"""Return first image in component folder."""
|
|
root, files = find_component_dir(component_name, pdk_root_for_request_project())
|
|
if root:
|
|
# Use the first common image asset in the component folder as its
|
|
# preview thumbnail.
|
|
for ext in ('.png', '.jpg', '.jpeg', '.svg'):
|
|
for f in files:
|
|
if f.lower().endswith(ext):
|
|
return send_from_directory(root, f)
|
|
return jsonify({"error": "No image found"}), 404
|
|
|
|
if __name__ == '__main__':
|
|
# Allow deployment scripts to choose host, port, and debug behavior through
|
|
# environment variables.
|
|
host = os.environ.get('MXPIC_HOST', '0.0.0.0')
|
|
port = int(os.environ.get('MXPIC_PORT', '3000'))
|
|
debug = os.environ.get('MXPIC_DEBUG', '0').lower() in {'1', 'true', 'yes'}
|
|
print(f"Starting mxpic EDA Server on http://{host}:{port}")
|
|
app.run(host=host, port=port, debug=debug, threaded=True)
|
|
|
|
|
|
|
|
|