269 lines
9.0 KiB
Python
269 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from app.core.config import get_settings
|
|
from app.models import (
|
|
Component,
|
|
ComponentProperty,
|
|
ComponentVariant,
|
|
DesignatorMapping,
|
|
DielMaterial,
|
|
PcbData,
|
|
Project,
|
|
ProjectParam,
|
|
Variant,
|
|
VariantProperty,
|
|
)
|
|
from app.services.altium_parser import AltiumParser, ProjectData, find_prjpcb, make_complex_string
|
|
from app.services.designators import DEFAULT_MAPPINGS
|
|
|
|
|
|
def project_dir(project_id: int) -> Path:
|
|
root = Path(get_settings().data_dir) / "projects" / str(project_id)
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
return root
|
|
|
|
|
|
def clear_parsed_data(db: Session, project: Project, keep_tables: bool = True) -> None:
|
|
"""Remove components/variants/params/pcb; optionally keep edited tables."""
|
|
comps = db.scalars(select(Component).where(Component.project_id == project.id)).all()
|
|
comp_ids = [c.id for c in comps]
|
|
if comp_ids:
|
|
db.execute(delete(VariantProperty).where(VariantProperty.component_id.in_(comp_ids)))
|
|
db.execute(delete(ComponentVariant).where(ComponentVariant.component_id.in_(comp_ids)))
|
|
db.execute(delete(ComponentProperty).where(ComponentProperty.component_id.in_(comp_ids)))
|
|
db.execute(delete(Component).where(Component.id.in_(comp_ids)))
|
|
db.execute(delete(Variant).where(Variant.project_id == project.id))
|
|
db.execute(delete(ProjectParam).where(ProjectParam.project_id == project.id))
|
|
pcb = db.scalar(select(PcbData).where(PcbData.project_id == project.id))
|
|
if pcb:
|
|
db.execute(delete(DielMaterial).where(DielMaterial.pcb_data_id == pcb.id))
|
|
db.delete(pcb)
|
|
if not keep_tables:
|
|
from app.models import (
|
|
PerechenRow,
|
|
SimpleListRow,
|
|
SpecificationPcbRow,
|
|
SpecificationRow,
|
|
VedomostRow,
|
|
)
|
|
|
|
for model in (PerechenRow, SpecificationPcbRow, SpecificationRow, VedomostRow, SimpleListRow):
|
|
db.execute(delete(model).where(model.project_id == project.id))
|
|
db.flush()
|
|
|
|
|
|
def ensure_default_mappings(db: Session, project: Project) -> None:
|
|
existing = db.scalars(
|
|
select(DesignatorMapping).where(DesignatorMapping.project_id == project.id)
|
|
).all()
|
|
if existing:
|
|
return
|
|
for prefix, (sing, plur) in DEFAULT_MAPPINGS.items():
|
|
db.add(
|
|
DesignatorMapping(
|
|
project_id=project.id,
|
|
prefix=prefix,
|
|
singular_name=sing,
|
|
plural_name=plur,
|
|
)
|
|
)
|
|
|
|
|
|
def save_project_data(db: Session, project: Project, data: ProjectData) -> None:
|
|
variant_objs: list[Variant] = []
|
|
for name in data.variant_names:
|
|
v = Variant(project_id=project.id, name=name)
|
|
db.add(v)
|
|
variant_objs.append(v)
|
|
db.flush()
|
|
|
|
# Flat props for expression resolution of project params
|
|
flat_props = []
|
|
for comp_props in data.components_list:
|
|
flat_props.extend(comp_props)
|
|
|
|
# Regular components
|
|
designator_to_component: dict[str, Component] = {}
|
|
for comp_props in data.components_list:
|
|
designator = ""
|
|
for p in comp_props:
|
|
if p.name.lower() == "designator":
|
|
designator = p.text
|
|
break
|
|
if not designator:
|
|
continue
|
|
if designator in designator_to_component:
|
|
continue
|
|
comp = Component(project_id=project.id, designator=designator)
|
|
db.add(comp)
|
|
db.flush()
|
|
for p in comp_props:
|
|
db.add(ComponentProperty(component_id=comp.id, key=p.name, value=p.text))
|
|
designator_to_component[designator] = comp
|
|
# link to all variants as fitted by default
|
|
for v in variant_objs:
|
|
db.add(ComponentVariant(component_id=comp.id, variant_id=v.id, is_fitted=True))
|
|
|
|
# Variant overrides (skip No Variations at index 0 of variant_names)
|
|
for i, variant_comps in enumerate(data.components_variant_list):
|
|
variant_index = i + 1
|
|
if variant_index >= len(variant_objs):
|
|
continue
|
|
variant = variant_objs[variant_index]
|
|
dnf_list = (
|
|
data.dnf_variant_designators_list[i]
|
|
if i < len(data.dnf_variant_designators_list)
|
|
else []
|
|
)
|
|
for comp_props in variant_comps:
|
|
designator = ""
|
|
for p in comp_props:
|
|
if p.name == "Designator":
|
|
designator = p.text
|
|
break
|
|
if not designator:
|
|
continue
|
|
comp = designator_to_component.get(designator)
|
|
if not comp:
|
|
continue
|
|
is_dnf = designator in dnf_list
|
|
link = db.scalar(
|
|
select(ComponentVariant).where(
|
|
ComponentVariant.component_id == comp.id,
|
|
ComponentVariant.variant_id == variant.id,
|
|
)
|
|
)
|
|
if link:
|
|
link.is_fitted = not is_dnf
|
|
for p in comp_props:
|
|
if p.name in ("Designator", "Kind"):
|
|
continue
|
|
db.add(
|
|
VariantProperty(
|
|
component_id=comp.id,
|
|
variant_id=variant.id,
|
|
key=p.name,
|
|
value=p.text,
|
|
)
|
|
)
|
|
|
|
# Project params
|
|
for variant_idx, params in enumerate(data.prj_params_variant_list):
|
|
for item in params:
|
|
if len(item) < 2:
|
|
continue
|
|
name, value = item[0], item[1]
|
|
value = make_complex_string(value, flat_props)
|
|
db.add(
|
|
ProjectParam(
|
|
project_id=project.id,
|
|
name=name,
|
|
value=value,
|
|
variant_name=data.variant_names[variant_idx]
|
|
if variant_idx < len(data.variant_names)
|
|
else "",
|
|
)
|
|
)
|
|
|
|
# PCB
|
|
if data.pcb_layer_count or data.pcb_diel_materials:
|
|
pcb = PcbData(project_id=project.id, layer_count=data.pcb_layer_count)
|
|
db.add(pcb)
|
|
db.flush()
|
|
for m in data.pcb_diel_materials:
|
|
db.add(
|
|
DielMaterial(
|
|
pcb_data_id=pcb.id,
|
|
name=m.name,
|
|
value=m.value,
|
|
height=m.height,
|
|
diel_type=m.diel_type,
|
|
layer_number=m.layer_number,
|
|
)
|
|
)
|
|
|
|
project.pcb_doc_name = Path(data.pcb_doc_file_name).name if data.pcb_doc_file_name else None
|
|
if "No Variations" in data.variant_names:
|
|
project.current_variant = "No Variations"
|
|
elif data.variant_names:
|
|
project.current_variant = data.variant_names[0]
|
|
|
|
|
|
def ingest_zip(
|
|
db: Session,
|
|
project: Project,
|
|
zip_bytes: bytes,
|
|
filename: str = "project.zip",
|
|
keep_tables: bool = True,
|
|
) -> Project:
|
|
settings = get_settings()
|
|
Path(settings.data_dir).mkdir(parents=True, exist_ok=True)
|
|
pdir = project_dir(project.id)
|
|
zip_path = pdir / filename
|
|
extract_path = pdir / "extract"
|
|
|
|
if extract_path.exists():
|
|
shutil.rmtree(extract_path)
|
|
extract_path.mkdir(parents=True, exist_ok=True)
|
|
zip_path.write_bytes(zip_bytes)
|
|
|
|
try:
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
zf.extractall(extract_path)
|
|
except zipfile.BadZipFile as e:
|
|
project.status = "error"
|
|
project.error_message = f"Invalid zip: {e}"
|
|
db.commit()
|
|
raise ValueError(project.error_message) from e
|
|
|
|
prj = find_prjpcb(extract_path)
|
|
if not prj:
|
|
project.status = "error"
|
|
project.error_message = "No .PrjPcb found in archive"
|
|
db.commit()
|
|
raise ValueError(project.error_message)
|
|
|
|
project.status = "parsing"
|
|
project.zip_path = str(zip_path)
|
|
project.extract_path = str(extract_path)
|
|
project.error_message = None
|
|
db.commit()
|
|
|
|
try:
|
|
clear_parsed_data(db, project, keep_tables=keep_tables)
|
|
parser = AltiumParser()
|
|
data = parser.parse_prjpcb(prj)
|
|
save_project_data(db, project, data)
|
|
ensure_default_mappings(db, project)
|
|
project.status = "ready"
|
|
db.commit()
|
|
db.refresh(project)
|
|
return project
|
|
except Exception as e:
|
|
project.status = "error"
|
|
project.error_message = str(e)
|
|
db.commit()
|
|
raise
|
|
|
|
|
|
def get_project_full(db: Session, project_id: int) -> Project | None:
|
|
return db.scalar(
|
|
select(Project)
|
|
.where(Project.id == project_id)
|
|
.options(
|
|
selectinload(Project.variants),
|
|
selectinload(Project.components).selectinload(Component.properties),
|
|
selectinload(Project.project_params),
|
|
selectinload(Project.pcb_data).selectinload(PcbData.diel_materials),
|
|
selectinload(Project.inscriptions),
|
|
selectinload(Project.designator_mappings),
|
|
)
|
|
)
|