More annotation added to the program

This commit is contained in:
2026-05-30 12:44:44 +08:00
parent b3f29398f0
commit bf223b52ac
22 changed files with 729 additions and 353 deletions
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+20 -3
View File
@@ -10,21 +10,24 @@ import os
from werkzeug.security import generate_password_hash
from datetime import datetime
# Save the database in the backend folder
# Store application data in the shared database folder so all backend modules
# use the same SQLite file regardless of their import path.
DB_FILE = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "database", "mxpic_data.db"))
def connect_db():
"""Open a SQLite connection with row-style access for application data queries."""
conn = sqlite3.connect(DB_FILE, timeout=20)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=20000")
return conn
def init_db():
"""Create the user, profile, and audit-log tables required by the backend."""
os.makedirs(os.path.dirname(DB_FILE), exist_ok=True)
conn = connect_db()
cursor = conn.cursor()
# Create Users Table
# Core account table used by login, profile, and role-based PDK access.
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -33,6 +36,8 @@ def init_db():
)
''')
# Audit log table used by dashboard activity history and backend action
# tracing.
cursor.execute('''
CREATE TABLE IF NOT EXISTS user_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -48,6 +53,8 @@ def init_db():
)
''')
# Lightweight migrations keep older local SQLite files compatible after
# profile, credit, occupation, and role fields were added.
cursor.execute("PRAGMA table_info(users)")
existing_columns = {row[1] for row in cursor.fetchall()}
migrations = {
@@ -66,7 +73,8 @@ def init_db():
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.
# Default local accounts let developers test manager/developer/user access
# without manually editing the database.
cursor.execute("SELECT * FROM users WHERE username = 'admin'")
if not cursor.fetchone():
test_hash = generate_password_hash("123456")
@@ -89,6 +97,7 @@ def init_db():
conn.close()
def get_user(username):
"""Fetch login credentials and account status for a username."""
conn = connect_db()
cursor = conn.cursor()
cursor.execute("SELECT id, username, password_hash, user_group FROM users WHERE username = ?", (username,))
@@ -97,6 +106,7 @@ def get_user(username):
return user
def get_user_profile(user_id):
"""Fetch editable profile details for an authenticated user."""
conn = connect_db()
cursor = conn.cursor()
cursor.execute(
@@ -108,6 +118,7 @@ def get_user_profile(user_id):
return user
def get_user_auth_by_id(user_id):
"""Fetch password and account metadata by user id."""
conn = connect_db()
cursor = conn.cursor()
cursor.execute("SELECT id, username, password_hash FROM users WHERE id = ?", (user_id,))
@@ -116,6 +127,7 @@ def get_user_auth_by_id(user_id):
return user
def update_user_occupation(user_id, occupation):
"""Persist a user profile occupation update."""
conn = connect_db()
cursor = conn.cursor()
cursor.execute("UPDATE users SET occupation = ? WHERE id = ?", (occupation, user_id))
@@ -123,6 +135,7 @@ def update_user_occupation(user_id, occupation):
conn.close()
def update_user_password(user_id, password):
"""Persist a newly hashed password for a user."""
conn = connect_db()
cursor = conn.cursor()
cursor.execute("UPDATE users SET password_hash = ? WHERE id = ?", (generate_password_hash(password), user_id))
@@ -130,6 +143,7 @@ def update_user_password(user_id, password):
conn.close()
def add_user_log(user_id, username, action, project=None, cell=None, detail=None, ip_address=None):
"""Record an auditable user action with optional project and cell context."""
conn = connect_db()
cursor = conn.cursor()
cursor.execute(
@@ -152,6 +166,7 @@ def add_user_log(user_id, username, action, project=None, cell=None, detail=None
conn.close()
def list_user_logs(user_id, limit=200):
"""Return recent audit-log entries for display in the account page."""
conn = connect_db()
cursor = conn.cursor()
cursor.execute(
@@ -169,5 +184,7 @@ def list_user_logs(user_id, limit=200):
return rows
if __name__ == "__main__":
# Allow this module to be run directly when a fresh local database needs to
# be initialized outside the Flask server.
init_db()
print("Database initialized successfully.")
+27
View File
@@ -17,6 +17,7 @@ from pdk_registry import PdkRegistry
@dataclass
class BuildResult:
"""Container for GDS build output paths, status details, and engine metadata."""
output_path: str
engine: str
cells_built: List[str] = field(default_factory=list)
@@ -35,6 +36,8 @@ def build_project_gds(
if not cells:
raise ValueError("No saved cell YAML files found for this project")
# Prefer the routed builder whenever it is available because it understands
# bundle links, anchor connections, and PDK-aware routing rules.
try:
return _build_with_mxpic_router(
project_dir,
@@ -50,9 +53,13 @@ def build_project_gds(
f"Router import failed: {router_error}"
) from router_error
# Placement-only projects can still be exported with local GDS engines when
# the routed builder is not installed.
registry = PdkRegistry(pdk_public_root, prefer_full_gds=prefer_full_gds)
os.makedirs(os.path.dirname(output_path), exist_ok=True)
# gdstk is the preferred fallback; Nazca remains a secondary fallback for
# environments where gdstk is not installed.
try:
return _build_with_gdstk(cells, output_path, registry)
except ImportError as gdstk_error:
@@ -72,6 +79,9 @@ def _build_with_mxpic_router(
technology_manifest_path: str,
prefer_full_gds: bool,
) -> BuildResult:
"""Delegate routed project GDS generation to the external mxpic_router package."""
# mxpic_router lives beside this repository during local development, so add
# that sibling checkout to sys.path only when it exists.
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)
@@ -93,6 +103,7 @@ def _build_with_mxpic_router(
def _load_project_cells(project_dir: str) -> Dict[str, dict]:
"""Load saved cell YAML documents from a project directory."""
cells = {}
for filename in sorted(os.listdir(project_dir)):
if not filename.lower().endswith((".yml", ".yaml")):
@@ -106,12 +117,14 @@ def _load_project_cells(project_dir: str) -> Dict[str, dict]:
def _ordered_cell_names(cells: Dict[str, dict]) -> List[str]:
"""Order cells so dependencies are built before cells that reference them."""
composites = [name for name, data in cells.items() if data.get("type") != "project"]
projects = [name for name, data in cells.items() if data.get("type") == "project"]
return composites + projects
def _cells_have_links(cells: Dict[str, dict]) -> bool:
"""Detect whether any saved cell contains bundle links that require routed building."""
for data in cells.values():
for bundle in (data.get("bundles") or {}).values():
if bundle.get("links"):
@@ -120,12 +133,15 @@ def _cells_have_links(cells: Dict[str, dict]) -> bool:
def _build_with_gdstk(cells: Dict[str, dict], output_path: str, registry: PdkRegistry) -> BuildResult:
"""Assemble a project GDS with gdstk when Nazca or routed building is unavailable."""
import gdstk
library = gdstk.Library()
built_cells = {}
warnings = []
# Build composite cells before project cells so project-level references can
# reuse cells already inserted into the same GDS library.
for cell_name in _ordered_cell_names(cells):
data = cells[cell_name]
gds_cell = library.new_cell(_safe_cell_name(cell_name, built_cells))
@@ -137,6 +153,8 @@ def _build_with_gdstk(cells: Dict[str, dict], output_path: str, registry: PdkReg
rotation = math.radians(_number(instance.get("rotation")))
child = built_cells.get(component)
if child is None:
# External components are resolved through the active PDK
# registry and imported as references.
asset = registry.resolve(component)
if not asset.gds_path:
warnings.append(f"Missing GDS for {instance_name}: {component}")
@@ -154,10 +172,13 @@ def _build_with_gdstk(cells: Dict[str, dict], output_path: str, registry: PdkReg
def _import_public_gds(gdstk, library, gds_path: str):
"""Import public PDK GDS geometry into the output library."""
source = gdstk.read_gds(gds_path)
top_cells = source.top_level()
if not top_cells:
raise ValueError(f"No top-level cell found in {gds_path}")
# Avoid adding duplicate cell names when multiple instances reference the
# same imported PDK component.
for source_cell in source.cells:
if _library_cell_by_name(library, source_cell.name) is None:
library.add(source_cell)
@@ -165,11 +186,14 @@ def _import_public_gds(gdstk, library, gds_path: str):
def _build_with_nazca(cells: Dict[str, dict], output_path: str, registry: PdkRegistry) -> BuildResult:
"""Assemble a project GDS with Nazca cells and PDK component placements."""
import nazca as nd
warnings = []
built_cells = {}
ordered_names = _ordered_cell_names(cells)
# Nazca cells are built in dependency order and then the final project cell
# is exported as the top-level GDS.
for cell_name in ordered_names:
data = cells[cell_name]
with nd.Cell(cell_name) as current_cell:
@@ -195,6 +219,7 @@ def _build_with_nazca(cells: Dict[str, dict], output_path: str, registry: PdkReg
def _safe_cell_name(name: str, existing: dict) -> str:
"""Generate a backend-safe unique cell name for GDS/Nazca libraries."""
base = "".join(ch if ch.isalnum() or ch in "._$" else "_" for ch in str(name)) or "cell"
candidate = base
counter = 1
@@ -206,6 +231,7 @@ def _safe_cell_name(name: str, existing: dict) -> str:
def _library_cell_by_name(library, name: str):
"""Find a cell object in a loaded layout library by name."""
for cell in library.cells:
if cell.name == name:
return cell
@@ -213,6 +239,7 @@ def _library_cell_by_name(library, name: str):
def _number(value, default=0.0) -> float:
"""Convert numeric YAML values to floats with a stable default."""
try:
if value is None or value == "":
return default
+18
View File
@@ -13,6 +13,8 @@ import yaml
def create_layout_svg_from_gds(yaml_content: str, output_path: str, pdk_registry, project_dir: str = None) -> str:
"""Create an SVG preview by placing real public _BB.gds cells from layout YAML."""
layout = yaml.safe_load(yaml_content) or {}
# Try gdstk first because it can write SVG directly; keep Nazca as a GDS
# placement fallback for environments where gdstk is unavailable.
try:
return _create_with_gdstk(layout, output_path, pdk_registry, project_dir)
except ImportError as gdstk_error:
@@ -26,6 +28,7 @@ def create_layout_svg_from_gds(yaml_content: str, output_path: str, pdk_registry
def _create_with_gdstk(layout: dict, output_path: str, pdk_registry, project_dir: Optional[str]) -> str:
"""Generate preview SVG geometry using gdstk import and placement APIs."""
import gdstk
library = gdstk.Library()
@@ -37,9 +40,12 @@ def _create_with_gdstk(layout: dict, output_path: str, pdk_registry, project_dir
def _build_gdstk_cell(gdstk, library, layout: dict, pdk_registry, project_dir: Optional[str], cell_cache: Dict):
"""Build or reuse a gdstk cell for a saved layout document."""
cell_name = _safe_cell_name(layout.get("name") or "layout", library)
top = library.new_cell(cell_name)
# Each saved instance becomes a GDS reference to either another project cell
# or a resolved PDK asset.
for instance_name, instance in (layout.get("instances") or {}).items():
component = str(instance.get("component") or "")
x = _number(instance.get("x"))
@@ -54,9 +60,12 @@ def _build_gdstk_cell(gdstk, library, layout: dict, pdk_registry, project_dir: O
def _resolve_child_cell(gdstk, library, component: str, pdk_registry, project_dir: Optional[str], cell_cache: Dict):
"""Resolve a placed child component from local cells or PDK assets."""
if component in cell_cache:
return cell_cache[component]
# Project-local composite cells are resolved before external PDK components
# so nested user-created cells can appear in preview output.
local_layout = _load_local_layout(component, project_dir)
if local_layout is not None:
child = _build_gdstk_cell(gdstk, library, local_layout, pdk_registry, project_dir, cell_cache)
@@ -72,10 +81,12 @@ def _resolve_child_cell(gdstk, library, component: str, pdk_registry, project_di
def _import_gds_cell(gdstk, library, gds_path: str):
"""Import a GDS file and return the first usable cell for placement."""
source = gdstk.read_gds(gds_path)
top_cells = source.top_level()
if not top_cells:
raise ValueError(f"No top-level cell found in {gds_path}")
# Reuse already-imported cells by name to keep the preview library compact.
for source_cell in source.cells:
if _library_cell_by_name(library, source_cell.name) is None:
library.add(source_cell)
@@ -83,9 +94,12 @@ def _import_gds_cell(gdstk, library, gds_path: str):
def _create_with_nazca(layout: dict, output_path: str, pdk_registry, project_dir: Optional[str]) -> str:
"""Generate preview SVG geometry using Nazca when gdstk is unavailable."""
import nazca as nd
png_path = os.path.splitext(output_path)[0] + ".gds"
# Nazca can place the same GDS references as the preview path, but this
# backend still requires gdstk for final SVG conversion.
with nd.Cell(str(layout.get("name") or "layout")) as top:
for instance_name, instance in (layout.get("instances") or {}).items():
component = str(instance.get("component") or "")
@@ -101,6 +115,7 @@ def _create_with_nazca(layout: dict, output_path: str, pdk_registry, project_dir
def _load_local_layout(component: str, project_dir: Optional[str]) -> Optional[dict]:
"""Load a project-local composite layout referenced by another cell."""
if not project_dir or "/" in component or "\\" in component or component == "generate_with_forge":
return None
for ext in (".yml", ".yaml"):
@@ -112,6 +127,7 @@ def _load_local_layout(component: str, project_dir: Optional[str]) -> Optional[d
def _safe_cell_name(name, library) -> str:
"""Generate a backend-safe unique cell name for GDS/Nazca libraries."""
base = "".join(ch if ch.isalnum() or ch in "._$" else "_" for ch in str(name)) or "layout"
candidate = base
counter = 1
@@ -122,6 +138,7 @@ def _safe_cell_name(name, library) -> str:
def _library_cell_by_name(library, name: str):
"""Find a cell object in a loaded layout library by name."""
for cell in library.cells:
if cell.name == name:
return cell
@@ -129,6 +146,7 @@ def _library_cell_by_name(library, name: str):
def _number(value, default=0.0) -> float:
"""Convert numeric YAML values to floats with a stable default."""
try:
if value is None or value == "":
return default
+12
View File
@@ -10,6 +10,8 @@ import shutil
import uuid
# Role names control which PDK tree the backend exposes for browsing, preview,
# and GDS export requests.
MANAGER_GROUP = "manager"
DEVELOPER_GROUP = "developers"
USER_GROUP = "user"
@@ -17,12 +19,16 @@ ALLOWED_GROUPS = {MANAGER_GROUP, DEVELOPER_GROUP, USER_GROUP}
def normalize_user_group(user_group: str) -> str:
"""Normalize a session user group into the supported PDK access scope."""
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:
"""Resolve the PDK library root that belongs to a user group."""
group = normalize_user_group(user_group)
# Managers may access the private atlas PDK tree; all other roles stay on
# the public PDK tree used for normal project editing.
if group == MANAGER_GROUP:
return os.path.abspath(os.environ.get(
"MXPIC_PDK_ATLAS_ROOT",
@@ -35,14 +41,17 @@ def pdk_root_for_group(user_group: str, repo_root: str) -> str:
def pdk_root_for_session(session_obj, repo_root: str) -> str:
"""Resolve the active PDK root from the current Flask session."""
return pdk_root_for_group(session_obj.get("user_group"), repo_root)
def prefer_full_gds_for_session(session_obj) -> bool:
"""Decide whether resolved PDK assets should prefer full GDS files."""
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]:
"""Create a unique export directory and filename for generated downloads."""
export_id = uuid.uuid4().hex
filename = f"{project_name}.gds"
export_dir = os.path.abspath(os.path.join(export_root, export_id))
@@ -51,9 +60,12 @@ def create_export_path(export_root: str, project_name: str) -> tuple[str, str, s
def cleanup_expired_exports(export_root: str, max_age_seconds: int = 86400) -> None:
"""Remove old export folders so temporary download storage stays bounded."""
if not os.path.isdir(export_root):
return
now = time.time()
# Each export is stored in its own UUID directory, so old folders can be
# removed independently without touching active project data.
for name in os.listdir(export_root):
path = os.path.join(export_root, name)
if not os.path.isdir(path):
+15
View File
@@ -13,6 +13,7 @@ import yaml
@dataclass
class PdkAsset:
"""Container describing the YAML and GDS assets resolved for a PDK component."""
component: str
yaml_path: Optional[str] = None
gds_path: Optional[str] = None
@@ -23,11 +24,13 @@ class PdkRegistry:
"""Resolve public PDK component names to metadata and public GDS assets."""
def __init__(self, public_root: str, prefer_full_gds: bool = False):
"""Store the active PDK root and cache resolved component assets."""
self.public_root = os.path.abspath(public_root)
self.prefer_full_gds = prefer_full_gds
self._asset_cache = {}
def resolve(self, component: str) -> PdkAsset:
"""Resolve YAML and GDS assets for a requested component key."""
key = (component or "").strip().replace("\\", "/").strip("/")
if not key:
return PdkAsset(component=component)
@@ -42,6 +45,9 @@ class PdkRegistry:
return asset
def _find_yaml(self, key: str) -> Optional[str]:
"""Locate the component YAML description file in the active PDK tree."""
# Try direct component paths first so saved YAML component references
# resolve without scanning the whole PDK tree.
direct = os.path.join(self.public_root, *key.split("/"))
candidates = []
if direct.lower().endswith((".yml", ".yaml")):
@@ -56,6 +62,8 @@ class PdkRegistry:
return os.path.abspath(candidate)
name = key.split("/")[-1]
# Fall back to a tree scan for older saved references that only stored
# the component folder name.
for root, dirs, files in os.walk(self.public_root):
if os.path.basename(root) == name:
for filename in files:
@@ -65,8 +73,11 @@ class PdkRegistry:
return None
def _find_gds(self, key: str, yaml_path: Optional[str]) -> Optional[str]:
"""Locate the best matching GDS asset for a component YAML or key."""
search_dir = os.path.dirname(yaml_path) if yaml_path else os.path.join(self.public_root, *key.split("/"))
name = key.split("/")[-1]
# Normal users prefer black-box GDS for fast previews; manager sessions
# can prefer full layout geometry for complete export.
if self.prefer_full_gds:
candidates = [
os.path.join(search_dir, f"{name}.gds"),
@@ -80,6 +91,8 @@ class PdkRegistry:
for candidate in candidates:
if self._inside_root(candidate) and os.path.exists(candidate):
return os.path.abspath(candidate)
# If the expected filename is missing, choose the first available GDS in
# the component folder while respecting the full-vs-BB preference.
if os.path.isdir(search_dir):
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")]
@@ -90,11 +103,13 @@ class PdkRegistry:
return None
def _load_yaml(self, yaml_path: Optional[str]) -> Optional[dict]:
"""Read a YAML component metadata file into a dictionary."""
if not yaml_path:
return None
with open(yaml_path, "r", encoding="utf-8") as file:
return yaml.safe_load(file) or {}
def _inside_root(self, path: str) -> bool:
"""Check that a candidate asset path remains inside the permitted PDK root."""
target = os.path.abspath(path)
return target == self.public_root or target.startswith(self.public_root + os.sep)
+7
View File
@@ -24,12 +24,16 @@ def create_routed_layout_svg(
layout = yaml.safe_load(yaml_content) or {}
cell_name = str(layout.get("name") or "layout")
# mxpic_router is kept as a sibling repository, so the backend adds it to
# sys.path only when the local checkout is available.
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)
# Build into a temporary GDS first, then convert the generated top cell into
# the SVG preview consumed by the canvas.
with tempfile.TemporaryDirectory(prefix="mxpic_routed_preview_") as temp_dir:
temp_gds = os.path.join(temp_dir, f"{cell_name}.gds")
build_project_gds(
@@ -49,7 +53,10 @@ def create_routed_layout_svg(
def layout_has_links(yaml_content: str) -> bool:
"""Detect whether a layout YAML document contains routed bundle links."""
layout = yaml.safe_load(yaml_content) or {}
# Any bundle link means preview generation must use the routed builder
# instead of simple component placement.
for bundle in (layout.get("bundles") or {}).values():
links = bundle.get("links") or []
if links:
+81 -5
View File
@@ -29,6 +29,8 @@ 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')
@@ -39,14 +41,17 @@ 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')
# Define where your new icons folder is located (adjust if it's placed elsewhere)
# Component/category icons are served from backend/icons for the library panel.
ICONS_DIR = os.path.join(BASE_DIR, 'icons')
#build layout save path
# 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(
@@ -56,6 +61,7 @@ app.config.update(
)
app.json.sort_keys = False
# Initialize the lightweight local database when the server process starts.
database.init_db()
@@ -68,8 +74,10 @@ def no_cache_response(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)
@@ -77,6 +85,7 @@ def login_required_json(view_func):
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()
@@ -84,6 +93,7 @@ def request_ip():
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)):
@@ -112,29 +122,37 @@ def safe_name(value, 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 []:
@@ -159,34 +177,43 @@ def write_route_points_sidecar(yaml_content, 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
@@ -199,15 +226,18 @@ def scoped_pdk_root_for_project(project_name):
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 {}
@@ -216,25 +246,32 @@ def read_project_meta(project_name):
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
# ... [Keep countSpaces and buildTree exactly as they are] ...
# --- 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:
@@ -260,6 +297,8 @@ 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']
@@ -280,13 +319,15 @@ def addCompsToTree(compMap):
return fresh_tree
# ... [Keep readCompYaml and Page Routes exactly as they are] ...
# 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):
@@ -304,11 +345,14 @@ def getIcon(category):
)
return Response(transparent_png, mimetype='image/png')
# ... [Keep existing API routes below] ...
# 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()
@@ -321,6 +365,7 @@ def readCompYaml(compName, comps_root=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:
@@ -387,6 +432,7 @@ def logout():
@app.route('/api/health')
def health_check():
"""Expose a small deployment health endpoint."""
return jsonify({"status": "ok", "service": "mxpic_eda"})
@@ -396,6 +442,8 @@ 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)
@@ -417,6 +465,7 @@ def list_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,
@@ -431,6 +480,9 @@ def get_technology_manifest(foundry, technology):
@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')
@@ -460,6 +512,7 @@ def account_profile():
@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 ''
@@ -478,6 +531,7 @@ def change_password():
@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'), '')
@@ -520,6 +574,8 @@ def list_projects():
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):
@@ -547,12 +603,15 @@ def list_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)):
@@ -618,6 +677,7 @@ def delete_project(project_name):
@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))
@@ -653,6 +713,7 @@ def rename_cell(project_name, cell_name):
@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')
@@ -670,6 +731,8 @@ def save_layout():
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,
@@ -699,6 +762,7 @@ def save_layout():
@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))
@@ -714,9 +778,11 @@ def get_layout_svg(project_name, cell_name):
@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):
@@ -751,9 +817,12 @@ def build_gds():
@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'):
@@ -769,6 +838,7 @@ def download_export(export_id, filename):
@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")
@@ -792,6 +862,8 @@ 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)
@@ -814,6 +886,8 @@ 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):
@@ -821,6 +895,8 @@ def getCompImg(component_name):
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'}
+7
View File
@@ -10,23 +10,30 @@ import yaml
class TechnologyManifestError(Exception):
"""Exception raised when a technology manifest cannot be found or parsed."""
pass
def technology_manifest_path(pdks_root: str, foundry: str, technology: str) -> str:
"""Build the expected path to a foundry/technology manifest YAML file."""
base = os.path.abspath(pdks_root)
path = os.path.abspath(os.path.join(base, foundry, technology, "technology.yml"))
# Keep user-provided foundry/technology names from escaping the configured
# PDK root.
if not path.startswith(base + os.sep):
raise TechnologyManifestError("Invalid technology path")
return path
def read_technology_manifest(pdks_root: str, foundry: str, technology: str) -> dict:
"""Load and validate the active technology manifest for the frontend."""
path = technology_manifest_path(pdks_root, foundry, technology)
if not os.path.exists(path):
raise TechnologyManifestError("technology manifest not generated; run mxpic_forge technology export workflow")
with open(path, "r", encoding="utf-8") as file:
manifest = yaml.safe_load(file) or {}
# The frontend route editor depends on both xsection definitions and global
# defaults being present before it can safely style or build links.
if not isinstance(manifest.get("xsections"), dict):
raise TechnologyManifestError("technology manifest is missing xsections")
if not isinstance(manifest.get("defaults"), dict):