CODEX revised with following function: 1. GDS building, 2. different user group with different authority.

This commit is contained in:
2026-05-28 20:35:49 +08:00
parent e6e9e13cf2
commit 1215bf978a
25 changed files with 439 additions and 196 deletions
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+10 -6
View File
@@ -48,6 +48,7 @@ def init_db():
"created_at": "ALTER TABLE users ADD COLUMN created_at TEXT",
"credits": "ALTER TABLE users ADD COLUMN credits INTEGER NOT NULL DEFAULT 0",
"occupation": "ALTER TABLE users ADD COLUMN occupation TEXT NOT NULL DEFAULT 'intern'",
"user_group": "ALTER TABLE users ADD COLUMN user_group TEXT NOT NULL DEFAULT 'user'",
}
for column, statement in migrations.items():
if column not in existing_columns:
@@ -55,14 +56,17 @@ def init_db():
now = datetime.utcnow().strftime("%Y-%m-%d")
cursor.execute("UPDATE users SET created_at = ? WHERE created_at IS NULL OR created_at = ''", (now,))
cursor.execute("UPDATE users SET user_group = 'manager' WHERE username = 'admin'")
cursor.execute("UPDATE users SET user_group = 'developers' WHERE username = 'engineer'")
cursor.execute("UPDATE users SET user_group = 'user' WHERE user_group IS NULL OR user_group = ''")
# Insert default users for local multi-account development.
cursor.execute("SELECT * FROM users WHERE username = 'admin'")
if not cursor.fetchone():
test_hash = generate_password_hash("123456")
cursor.execute(
"INSERT INTO users (username, password_hash, created_at, credits, occupation) VALUES (?, ?, ?, ?, ?)",
("admin", test_hash, now, 0, "principle engineer")
"INSERT INTO users (username, password_hash, created_at, credits, occupation, user_group) VALUES (?, ?, ?, ?, ?, ?)",
("admin", test_hash, now, 0, "principle engineer", "manager")
)
print("Test user created. Username: admin | Password: 123456")
@@ -70,8 +74,8 @@ def init_db():
if not cursor.fetchone():
test_hash = generate_password_hash("123456")
cursor.execute(
"INSERT INTO users (username, password_hash, created_at, credits, occupation) VALUES (?, ?, ?, ?, ?)",
("engineer", test_hash, now, 0, "junior engineer")
"INSERT INTO users (username, password_hash, created_at, credits, occupation, user_group) VALUES (?, ?, ?, ?, ?, ?)",
("engineer", test_hash, now, 0, "junior engineer", "developers")
)
print("Second test user created. Username: engineer | Password: 123456")
@@ -81,7 +85,7 @@ def init_db():
def get_user(username):
conn = connect_db()
cursor = conn.cursor()
cursor.execute("SELECT id, username, password_hash FROM users WHERE username = ?", (username,))
cursor.execute("SELECT id, username, password_hash, user_group FROM users WHERE username = ?", (username,))
user = cursor.fetchone()
conn.close()
return user
@@ -90,7 +94,7 @@ def get_user_profile(user_id):
conn = connect_db()
cursor = conn.cursor()
cursor.execute(
"SELECT id, username, created_at, credits, occupation FROM users WHERE id = ?",
"SELECT id, username, created_at, credits, occupation, user_group FROM users WHERE id = ?",
(user_id,)
)
user = cursor.fetchone()
+59 -2
View File
@@ -1,5 +1,6 @@
import math
import os
import sys
from dataclasses import dataclass, field
from typing import Dict, List
@@ -16,13 +17,34 @@ class BuildResult:
warnings: List[str] = field(default_factory=list)
def build_project_gds(project_dir: str, output_path: str, pdk_public_root: str) -> BuildResult:
def build_project_gds(
project_dir: str,
output_path: str,
pdk_public_root: str,
technology_manifest_path: str = None,
prefer_full_gds: bool = False,
) -> BuildResult:
"""Build a hierarchical project GDS from saved cell YAML files."""
cells = _load_project_cells(project_dir)
if not cells:
raise ValueError("No saved cell YAML files found for this project")
registry = PdkRegistry(pdk_public_root)
try:
return _build_with_mxpic_router(
project_dir,
output_path,
pdk_public_root,
technology_manifest_path,
prefer_full_gds,
)
except ImportError as router_error:
if _cells_have_links(cells):
raise RuntimeError(
"Routed Build GDS requires mxpic_router, nazca, and mxpic_forge when layout links are present. "
f"Router import failed: {router_error}"
) from router_error
registry = PdkRegistry(pdk_public_root, prefer_full_gds=prefer_full_gds)
os.makedirs(os.path.dirname(output_path), exist_ok=True)
try:
@@ -37,6 +59,33 @@ def build_project_gds(project_dir: str, output_path: str, pdk_public_root: str)
) from nazca_error
def _build_with_mxpic_router(
project_dir: str,
output_path: str,
pdk_root: str,
technology_manifest_path: str,
prefer_full_gds: bool,
) -> BuildResult:
router_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "mxpic_router"))
if os.path.isdir(router_root) and router_root not in sys.path:
sys.path.insert(0, router_root)
from mxpic_router import build_project_gds as build_routed_project_gds
result = build_routed_project_gds(
project_dir=project_dir,
output_path=output_path,
pdk_root=pdk_root,
technology_manifest_path=technology_manifest_path,
prefer_full_gds=prefer_full_gds,
)
return BuildResult(
output_path=result.get("output_path", output_path),
engine=result.get("engine", "mxpic_router"),
cells_built=result.get("cells_built", []),
warnings=result.get("warnings", []),
)
def _load_project_cells(project_dir: str) -> Dict[str, dict]:
cells = {}
for filename in sorted(os.listdir(project_dir)):
@@ -56,6 +105,14 @@ def _ordered_cell_names(cells: Dict[str, dict]) -> List[str]:
return composites + projects
def _cells_have_links(cells: Dict[str, dict]) -> bool:
for data in cells.values():
for bundle in (data.get("bundles") or {}).values():
if bundle.get("links"):
return True
return False
def _build_with_gdstk(cells: Dict[str, dict], output_path: str, registry: PdkRegistry) -> BuildResult:
import gdstk
+60
View File
@@ -0,0 +1,60 @@
import os
import time
import shutil
import uuid
MANAGER_GROUP = "manager"
DEVELOPER_GROUP = "developers"
USER_GROUP = "user"
ALLOWED_GROUPS = {MANAGER_GROUP, DEVELOPER_GROUP, USER_GROUP}
def normalize_user_group(user_group: str) -> str:
group = (user_group or USER_GROUP).strip().lower()
return group if group in ALLOWED_GROUPS else USER_GROUP
def pdk_root_for_group(user_group: str, repo_root: str) -> str:
group = normalize_user_group(user_group)
if group == MANAGER_GROUP:
return os.path.abspath(os.environ.get(
"MXPIC_PDK_ATLAS_ROOT",
os.path.join(repo_root, "opt_pdk_atlas", "foundries"),
))
return os.path.abspath(os.environ.get(
"MXPIC_PDK_PUBLIC_ROOT",
os.path.join(repo_root, "opt_pdk_public", "foundries"),
))
def pdk_root_for_session(session_obj, repo_root: str) -> str:
return pdk_root_for_group(session_obj.get("user_group"), repo_root)
def prefer_full_gds_for_session(session_obj) -> bool:
return normalize_user_group(session_obj.get("user_group")) == MANAGER_GROUP
def create_export_path(export_root: str, project_name: str) -> tuple[str, str, str]:
export_id = uuid.uuid4().hex
filename = f"{project_name}.gds"
export_dir = os.path.abspath(os.path.join(export_root, export_id))
os.makedirs(export_dir, exist_ok=True)
return export_id, filename, os.path.join(export_dir, filename)
def cleanup_expired_exports(export_root: str, max_age_seconds: int = 86400) -> None:
if not os.path.isdir(export_root):
return
now = time.time()
for name in os.listdir(export_root):
path = os.path.join(export_root, name)
if not os.path.isdir(path):
continue
try:
age = now - os.path.getmtime(path)
if age > max_age_seconds:
shutil.rmtree(path, ignore_errors=True)
except OSError:
continue
+18 -11
View File
@@ -16,8 +16,9 @@ class PdkAsset:
class PdkRegistry:
"""Resolve public PDK component names to metadata and public GDS assets."""
def __init__(self, public_root: str):
def __init__(self, public_root: str, prefer_full_gds: bool = False):
self.public_root = os.path.abspath(public_root)
self.prefer_full_gds = prefer_full_gds
self._asset_cache = {}
def resolve(self, component: str) -> PdkAsset:
@@ -60,20 +61,26 @@ class PdkRegistry:
def _find_gds(self, key: str, yaml_path: Optional[str]) -> Optional[str]:
search_dir = os.path.dirname(yaml_path) if yaml_path else os.path.join(self.public_root, *key.split("/"))
name = key.split("/")[-1]
candidates = [
os.path.join(search_dir, f"{name}_BB.gds"),
os.path.join(search_dir, f"{name}.gds"),
]
if self.prefer_full_gds:
candidates = [
os.path.join(search_dir, f"{name}.gds"),
os.path.join(search_dir, f"{name}_BB.gds"),
]
else:
candidates = [
os.path.join(search_dir, f"{name}_BB.gds"),
os.path.join(search_dir, f"{name}.gds"),
]
for candidate in candidates:
if self._inside_root(candidate) and os.path.exists(candidate):
return os.path.abspath(candidate)
if os.path.isdir(search_dir):
for filename in sorted(os.listdir(search_dir)):
if filename.lower().endswith("_bb.gds"):
return os.path.join(search_dir, filename)
for filename in sorted(os.listdir(search_dir)):
if filename.lower().endswith(".gds"):
return os.path.join(search_dir, filename)
gds_files = sorted(filename for filename in os.listdir(search_dir) if filename.lower().endswith(".gds"))
full_files = [filename for filename in gds_files if not filename.lower().endswith("_bb.gds")]
bb_files = [filename for filename in gds_files if filename.lower().endswith("_bb.gds")]
ordered = (full_files + bb_files) if self.prefer_full_gds else (bb_files + full_files)
if ordered:
return os.path.join(search_dir, ordered[0])
return None
def _load_yaml(self, yaml_path: Optional[str]) -> Optional[dict]:
+51
View File
@@ -0,0 +1,51 @@
import os
import sys
import tempfile
import yaml
def create_routed_layout_svg(
yaml_content: str,
output_path: str,
pdk_root: str,
project_dir: str,
technology_manifest_path: str = None,
prefer_full_gds: bool = False,
) -> str:
"""Create an SVG preview from routed GDS geometry generated by mxpic_router."""
import gdstk
layout = yaml.safe_load(yaml_content) or {}
cell_name = str(layout.get("name") or "layout")
router_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "mxpic_router"))
if os.path.isdir(router_root) and router_root not in sys.path:
sys.path.insert(0, router_root)
from mxpic_router import build_project_gds
os.makedirs(os.path.dirname(output_path), exist_ok=True)
with tempfile.TemporaryDirectory(prefix="mxpic_routed_preview_") as temp_dir:
temp_gds = os.path.join(temp_dir, f"{cell_name}.gds")
build_project_gds(
project_dir=project_dir,
output_path=temp_gds,
pdk_root=pdk_root,
technology_manifest_path=technology_manifest_path,
prefer_full_gds=prefer_full_gds,
target_cell_name=cell_name,
)
library = gdstk.read_gds(temp_gds)
top_cells = library.top_level()
if not top_cells:
raise RuntimeError("Routed preview GDS has no top-level cell")
top_cells[0].write_svg(output_path)
return output_path
def layout_has_links(yaml_content: str) -> bool:
layout = yaml.safe_load(yaml_content) or {}
for bundle in (layout.get("bundles") or {}).values():
links = bundle.get("links") or []
if links:
return True
return False
+81 -13
View File
@@ -13,6 +13,13 @@ 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 ---
@@ -26,13 +33,12 @@ PDK_PUBLIC_ROOT = os.path.abspath(os.environ.get(
))
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')
COMPS_ROOT = os.environ.get('MXPIC_COMPONENT_ROOT', PDK_PUBLIC_ROOT)
# Define where your new icons folder is located (adjust if it's placed elsewhere)
ICONS_DIR = os.path.join(BASE_DIR, 'icons')
#build layout save path
DATABASE_ROOT = os.path.abspath(os.path.join(BASE_DIR, '..', 'database'))
PDK_REGISTRY = PdkRegistry(PDK_PUBLIC_ROOT)
EXPORT_ROOT = os.path.abspath(os.path.join(DATABASE_ROOT, '_exports'))
app = Flask(__name__, template_folder=FRONTEND_DIR, static_folder=FRONTEND_DIR)
@@ -120,6 +126,25 @@ def project_gds_path(project_name):
return os.path.join(project_root(project_name), f"{safe_name(project_name, 'project_1')}.gds")
def technology_manifest_path_for_project(project_name):
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"))
if path.startswith(EDA_PDK_ROOT + os.sep) and os.path.exists(path):
return path
return None
def current_pdk_root():
return pdk_root_for_session(session, REPO_ROOT)
def current_pdk_registry():
return PdkRegistry(current_pdk_root(), prefer_full_gds=prefer_full_gds_for_session(session))
def project_meta_path(project_name):
return os.path.join(project_root(project_name), ".project.json")
@@ -220,9 +245,10 @@ def getIcon(category):
# ... [Keep existing API routes below] ...
def readCompYaml(compName):
def readCompYaml(compName, comps_root=None):
"""Load YAML from component folder."""
for root, dirs, files in os.walk(COMPS_ROOT):
search_root = comps_root or current_pdk_root()
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')]
@@ -252,6 +278,7 @@ def login():
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:
@@ -355,6 +382,7 @@ def account_profile():
"created_at": profile[2],
"credits": profile[3] or 0,
"occupation": profile[4] or "intern",
"user_group": profile[5] or "user",
"occupations": sorted(occupations)
})
@@ -568,7 +596,17 @@ def save_layout():
f.write(content)
svg_path = cell_svg_path(project, cell)
create_layout_svg_from_gds(content, svg_path, pdk_registry=PDK_REGISTRY, project_dir=project_root(project))
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({
@@ -605,11 +643,18 @@ def build_gds():
data = request.get_json(silent=True) or {}
project = safe_name(data.get('project'), 'project_1')
try:
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
output_path = project_gds_path(project)
result = build_project_gds(project_dir, output_path, PDK_PUBLIC_ROOT)
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,
@@ -619,7 +664,8 @@ def build_gds():
"message": "GDS built",
"project": project,
"path": result.output_path,
"gds_url": url_for('get_project_gds', project_name=project, filename=os.path.basename(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
@@ -628,6 +674,24 @@ def build_gds():
return jsonify({"error": str(e)}), 500
@app.route('/api/exports/<export_id>/<filename>')
@login_required_json
def download_export(export_id, filename):
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))
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):
@@ -649,28 +713,32 @@ def get_project_gds(project_name, filename):
# --- API ROUTES (Library & Components) ---
@app.route('/api/library')
@login_required_json
def getLib():
"""Get library structure."""
# tree = buildTree(YML_PATH)
if os.path.isdir(COMPS_ROOT):
compMap = findComps(COMPS_ROOT)
comps_root = current_pdk_root()
fresh_tree = {}
if os.path.isdir(comps_root):
compMap = findComps(comps_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)
data = readCompYaml(component_name, current_pdk_root())
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."""
for root, dirs, files in os.walk(COMPS_ROOT):
for root, dirs, files in os.walk(current_pdk_root()):
if os.path.basename(root) == component_name:
dirs.clear()
for ext in ('.png', '.jpg', '.jpeg', '.svg'):