459 lines
17 KiB
Python
459 lines
17 KiB
Python
"""Altium .PrjPcb / .SchDoc / .PcbDoc parser (ported from desktop AltiumParser)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
|
||
def sanitize_text(value: str | None) -> str:
|
||
"""Remove NUL bytes — PostgreSQL text fields reject them; SchDoc may contain them."""
|
||
if not value:
|
||
return ""
|
||
return value.replace("\x00", "")
|
||
|
||
|
||
@dataclass
|
||
class ComponentProperty:
|
||
name: str
|
||
text: str
|
||
|
||
def __post_init__(self) -> None:
|
||
self.name = sanitize_text(self.name)
|
||
self.text = sanitize_text(self.text)
|
||
|
||
|
||
@dataclass
|
||
class DielProperty:
|
||
name: str
|
||
value: str
|
||
height: float
|
||
diel_type: int
|
||
layer_number: int
|
||
|
||
def __post_init__(self) -> None:
|
||
self.name = sanitize_text(self.name)
|
||
self.value = sanitize_text(self.value)
|
||
|
||
|
||
@dataclass
|
||
class ProjectData:
|
||
variant_names: list[str] = field(default_factory=lambda: ["No Variations"])
|
||
components_list: list[list[ComponentProperty]] = field(default_factory=list)
|
||
components_variant_list: list[list[list[ComponentProperty]]] = field(default_factory=list)
|
||
components_prop_variant_list: list[list[ComponentProperty]] = field(default_factory=list)
|
||
component_variant_prop_list: list[ComponentProperty] = field(default_factory=list)
|
||
dnf_designators_list: list[str] = field(default_factory=list)
|
||
dnf_variant_designators_list: list[list[str]] = field(default_factory=list)
|
||
fitted_designators_list: list[str] = field(default_factory=list)
|
||
fitted_variant_designators_list: list[list[str]] = field(default_factory=list)
|
||
prj_params_variant_list: list[list[list[str]]] = field(default_factory=list)
|
||
pcb_layer_count: int = 0
|
||
pcb_diel_materials: list[DielProperty] = field(default_factory=list)
|
||
pcb_doc_file_name: str = ""
|
||
is_waiting_variant_description: bool = False
|
||
current_variant_number: int = 0
|
||
|
||
|
||
_MOJIBAKE_RE = re.compile(r"[\u0420\u0421][\u0400-\u045f\u2018\u2019\u2116]")
|
||
_RU_PARAM_RE = re.compile(
|
||
r"Вид_документа|Дата_изменения|Код_документа|Схема|документ|изменения|платы",
|
||
re.I,
|
||
)
|
||
_CYR_WORD_RE = re.compile(r"[А-Яа-яЁё]{4,}")
|
||
|
||
|
||
def _encoding_score(text: str) -> float:
|
||
"""Higher = more likely correct decoding for Altium Russian project files."""
|
||
if not text:
|
||
return 0.0
|
||
sample = text[:20000]
|
||
score = 0.0
|
||
score -= sample.count("\ufffd") * 10
|
||
score -= len(_MOJIBAKE_RE.findall(sample)) * 0.4
|
||
score += len(_CYR_WORD_RE.findall(sample)) * 1.5
|
||
if _RU_PARAM_RE.search(sample):
|
||
score += 40
|
||
# UTF-8 misread as cp1251 often contains isolated Р' / Р· patterns
|
||
if "Р'" in sample or "Р·" in sample or "Р°" in sample:
|
||
score -= 25
|
||
return score
|
||
|
||
|
||
def decode_altium_bytes(raw: bytes) -> str:
|
||
"""Pick UTF-8 or CP1251 — newer Altium saves UTF-8, older uses Windows-1251."""
|
||
if raw.startswith(b"\xef\xbb\xbf"):
|
||
return raw[3:].decode("utf-8")
|
||
|
||
candidates: list[tuple[float, str]] = []
|
||
for enc in ("utf-8", "cp1251"):
|
||
try:
|
||
text = raw.decode(enc)
|
||
except UnicodeDecodeError:
|
||
continue
|
||
candidates.append((_encoding_score(text), text))
|
||
|
||
if not candidates:
|
||
return raw.decode("cp1251", errors="replace")
|
||
|
||
candidates.sort(key=lambda item: item[0], reverse=True)
|
||
return candidates[0][1]
|
||
|
||
|
||
def _read_text_lines(path: Path) -> list[str]:
|
||
text = decode_altium_bytes(path.read_bytes())
|
||
return [line.rstrip("\r\n") for line in text.splitlines()]
|
||
|
||
|
||
def _read_pipe_chunks(path: Path) -> list[str]:
|
||
"""SchDoc/PcbDoc are binary; desktop always reads as CP1251 text lines."""
|
||
raw = path.read_bytes()
|
||
text = raw.decode("cp1251", errors="replace")
|
||
return [line for line in text.splitlines() if line.strip()]
|
||
|
||
|
||
def _resolve_document_path(prj_path: Path, rel: str) -> Path | None:
|
||
rel = rel.strip().strip('"').replace("\\", "/")
|
||
candidate = (prj_path.parent / rel).resolve()
|
||
if candidate.exists():
|
||
return candidate
|
||
name = Path(rel).name
|
||
if not name:
|
||
return None
|
||
# Search near project file, then whole extract tree
|
||
for base in (prj_path.parent, prj_path.parent.parent):
|
||
matches = list(base.rglob(name))
|
||
if matches:
|
||
return matches[0].resolve()
|
||
return None
|
||
|
||
|
||
class AltiumParser:
|
||
def __init__(self) -> None:
|
||
self.last_error = ""
|
||
self.pcb_doc_file_name = ""
|
||
|
||
def parse_prjpcb(self, filename: str | Path) -> ProjectData:
|
||
path = Path(filename)
|
||
if not path.exists():
|
||
self.last_error = f"File not found: {path}"
|
||
raise FileNotFoundError(self.last_error)
|
||
|
||
data = ProjectData()
|
||
data.prj_params_variant_list = [[]] # index 0 = No Variations
|
||
|
||
lines = _read_text_lines(path)
|
||
prev = ""
|
||
prev_prev = ""
|
||
for line in lines:
|
||
prj = line.strip()
|
||
self._parse_project_variant_section(prj, data)
|
||
self._parse_component_variations(prj, data)
|
||
self._parse_project_parameters(prev_prev, prev, prj, data)
|
||
self._parse_schdoc_files(prj, path, data)
|
||
self._parse_pcbdoc_files(prj, path, data)
|
||
prev_prev, prev = prev, prj
|
||
|
||
# finalize last variant buffers if any
|
||
if (
|
||
data.component_variant_prop_list
|
||
or data.components_prop_variant_list
|
||
or data.dnf_designators_list
|
||
or data.fitted_designators_list
|
||
):
|
||
self._finalize_current_variant(data)
|
||
|
||
self.pcb_doc_file_name = data.pcb_doc_file_name
|
||
return data
|
||
|
||
def _finalize_current_variant(self, data: ProjectData) -> None:
|
||
if data.component_variant_prop_list:
|
||
data.components_prop_variant_list.append(list(data.component_variant_prop_list))
|
||
data.component_variant_prop_list.clear()
|
||
if data.components_prop_variant_list:
|
||
data.components_variant_list.append(list(data.components_prop_variant_list))
|
||
data.components_prop_variant_list.clear()
|
||
data.dnf_variant_designators_list.append(list(data.dnf_designators_list))
|
||
data.fitted_variant_designators_list.append(list(data.fitted_designators_list))
|
||
data.dnf_designators_list.clear()
|
||
data.fitted_designators_list.clear()
|
||
|
||
def _parse_project_variant_section(self, prj: str, data: ProjectData) -> None:
|
||
upper = prj.upper()
|
||
if len(prj) > 15 and upper.startswith("[PROJECTVARIANT"):
|
||
data.component_variant_prop_list.clear()
|
||
data.is_waiting_variant_description = True
|
||
clean = prj.replace("]", "")
|
||
suffix = clean[15:].strip()
|
||
data.current_variant_number = int(suffix) if suffix.isdigit() else len(data.variant_names)
|
||
while len(data.prj_params_variant_list) <= data.current_variant_number:
|
||
data.prj_params_variant_list.append([])
|
||
return
|
||
if len(prj) > 18 and upper.startswith("PARAMVARIATIONCOUNT"):
|
||
self._finalize_current_variant(data)
|
||
return
|
||
if data.is_waiting_variant_description and len(prj) > 12 and upper.startswith("DESCRIPTION="):
|
||
name = prj[12:].strip()
|
||
data.is_waiting_variant_description = False
|
||
if name and name not in data.variant_names:
|
||
data.variant_names.append(name)
|
||
|
||
def _parse_variation_designator(
|
||
self, parts: list[str], values: list[str], data: ProjectData
|
||
) -> None:
|
||
if data.component_variant_prop_list:
|
||
data.components_prop_variant_list.append(list(data.component_variant_prop_list))
|
||
data.component_variant_prop_list.clear()
|
||
|
||
designator = values[2] if len(values) > 2 else ""
|
||
if not designator:
|
||
return
|
||
data.component_variant_prop_list.append(ComponentProperty("Designator", designator))
|
||
|
||
kind = "0"
|
||
if len(parts) > 2:
|
||
kind_part = parts[2]
|
||
if "=" in kind_part:
|
||
k, v = kind_part.split("=", 1)
|
||
if k.upper() == "KIND":
|
||
kind = v
|
||
data.component_variant_prop_list.append(ComponentProperty("Kind", kind))
|
||
if kind == "1":
|
||
data.dnf_designators_list.append(designator)
|
||
else:
|
||
data.fitted_designators_list.append(designator)
|
||
|
||
def _parse_param_variation(
|
||
self, parts: list[str], values: list[str], data: ProjectData
|
||
) -> None:
|
||
if len(values) <= 2:
|
||
return
|
||
prop_name = values[2]
|
||
prop_text = ""
|
||
if len(parts) > 1:
|
||
idx = parts[1].find("=")
|
||
prop_text = parts[1][idx + 1 :] if idx >= 0 else parts[1]
|
||
data.component_variant_prop_list.append(ComponentProperty(prop_name, prop_text))
|
||
|
||
def _parse_component_variations(self, prj: str, data: ProjectData) -> None:
|
||
if len(prj) <= 9:
|
||
return
|
||
parts = prj.split("|")
|
||
if not parts:
|
||
return
|
||
values = parts[0].split("=")
|
||
if len(values) <= 2:
|
||
return
|
||
|
||
key0 = values[0].upper()
|
||
key1 = values[1].upper() if len(values) > 1 else ""
|
||
|
||
if len(values[0]) >= 9 and key0[:9] == "VARIATION" and key1[:10] == "DESIGNATOR":
|
||
self._parse_variation_designator(parts, values, data)
|
||
return
|
||
|
||
if len(values[0]) >= 14 and key0[:14] == "PARAMVARIATION" and key1 == "PARAMETERNAME":
|
||
self._parse_param_variation(parts, values, data)
|
||
|
||
def _parse_project_parameters(
|
||
self, prev_prev: str, prev: str, prj: str, data: ProjectData
|
||
) -> None:
|
||
if not prev_prev.upper().startswith("[PARAMETER"):
|
||
return
|
||
if not prev.upper().startswith("NAME="):
|
||
return
|
||
if not prj.upper().startswith("VALUE="):
|
||
return
|
||
name = prev[5:]
|
||
value = prj[6:]
|
||
idx = data.current_variant_number if "_" in prev_prev else 0
|
||
while len(data.prj_params_variant_list) <= idx:
|
||
data.prj_params_variant_list.append([])
|
||
data.prj_params_variant_list[idx].append([name, value])
|
||
|
||
def _parse_schdoc_files(self, prj: str, prj_path: Path, data: ProjectData) -> None:
|
||
if len(prj) <= 13:
|
||
return
|
||
if prj[:13].upper() != "DOCUMENTPATH=":
|
||
return
|
||
if prj[-6:].upper() != "SCHDOC":
|
||
return
|
||
rel = prj[13:]
|
||
sch_path = _resolve_document_path(prj_path, rel)
|
||
if sch_path:
|
||
before = len(data.components_list)
|
||
self._parse_schdoc_file(sch_path, data.components_list)
|
||
if len(data.components_list) == before:
|
||
self._parse_schdoc_file_fallback(sch_path, data.components_list)
|
||
|
||
def _parse_pcbdoc_files(self, prj: str, prj_path: Path, data: ProjectData) -> None:
|
||
if len(prj) <= 13:
|
||
return
|
||
if prj[:13].upper() != "DOCUMENTPATH=":
|
||
return
|
||
if prj[-6:].upper() != "PCBDOC":
|
||
return
|
||
rel = prj[13:]
|
||
pcb_path = _resolve_document_path(prj_path, rel)
|
||
if pcb_path and self._parse_pcbdoc_file(pcb_path, data):
|
||
data.pcb_doc_file_name = str(pcb_path)
|
||
|
||
def _parse_schdoc_parts(
|
||
self, parts: list[str], components_list: list[list[ComponentProperty]]
|
||
) -> None:
|
||
is_component = False
|
||
is_no_bom = False
|
||
is_first_part = True
|
||
component_props: list[ComponentProperty] = []
|
||
|
||
def flush() -> None:
|
||
nonlocal component_props, is_no_bom, is_first_part
|
||
if not is_no_bom and component_props and is_first_part:
|
||
components_list.append(list(component_props))
|
||
component_props = []
|
||
is_no_bom = False
|
||
is_first_part = True
|
||
|
||
for i in range(len(parts) - 1):
|
||
part = parts[i]
|
||
up = part.upper()
|
||
|
||
if is_component:
|
||
if len(part) >= 23 and up.startswith("COMPONENTKINDVERSION2=5"):
|
||
is_no_bom = True
|
||
if len(part) >= 15 and up.startswith("CURRENTPARTID="):
|
||
if up != "CURRENTPARTID=1" or len(part) > 15:
|
||
is_first_part = False
|
||
nxt = parts[i + 1]
|
||
if (
|
||
len(part) > 5
|
||
and len(nxt) > 5
|
||
and up.startswith("TEXT=")
|
||
and nxt.upper().startswith("NAME=")
|
||
):
|
||
if not is_no_bom:
|
||
component_props.append(ComponentProperty(nxt[5:], part[5:]))
|
||
if len(part) >= 8 and (
|
||
up.startswith("HEADER=") or (up == "RECORD=1" and len(part) == 8)
|
||
):
|
||
flush()
|
||
is_component = False
|
||
|
||
if len(part) == 8 and up == "RECORD=1":
|
||
is_component = True
|
||
|
||
if is_component:
|
||
flush()
|
||
|
||
def _parse_schdoc_file_fallback(
|
||
self, path: Path, components_list: list[list[ComponentProperty]]
|
||
) -> None:
|
||
raw = path.read_bytes()
|
||
text = raw.decode("cp1251", errors="replace")
|
||
self._parse_schdoc_parts(text.split("|"), components_list)
|
||
|
||
def _parse_schdoc_file(self, path: Path, components_list: list[list[ComponentProperty]]) -> None:
|
||
lines = _read_pipe_chunks(path)
|
||
for line in lines:
|
||
self._parse_schdoc_parts(line.split("|"), components_list)
|
||
|
||
def _parse_pcbdoc_file(self, path: Path, data: ProjectData) -> bool:
|
||
lines = _read_pipe_chunks(path)
|
||
signal_layerset: Optional[str] = None
|
||
layer_count = 0
|
||
materials: list[DielProperty] = []
|
||
|
||
layerset_re = re.compile(r"LAYERSET(\d+)NAME=&Signal Layers", re.I)
|
||
v9_name_re = re.compile(r"V9_STACK_LAYER(\d+)_NAME=", re.I)
|
||
v9_diel_re = re.compile(r"V9_STACK_LAYER(\d+)_DIELTYPE=(\d+)", re.I)
|
||
v8_diel_re = re.compile(r"LAYER_V8_(\d+)_DIELTYPE=(\d+)", re.I)
|
||
|
||
found_layerset = False
|
||
for line in lines:
|
||
m = layerset_re.search(line)
|
||
if m:
|
||
signal_layerset = m.group(1)
|
||
found_layerset = True
|
||
|
||
for line in lines:
|
||
parts = line.split("|")
|
||
joined = line
|
||
if signal_layerset:
|
||
key = f"LAYERSET{signal_layerset}LAYERS="
|
||
for part in parts:
|
||
if part.upper().startswith(key.upper()):
|
||
layers = part.split("=", 1)[1].split(",")
|
||
layer_count = sum(1 for x in layers if x.strip() and x.strip() != "MultiLayer")
|
||
|
||
for m in v9_name_re.finditer(joined):
|
||
n = int(m.group(1))
|
||
layer_count = max(layer_count, n + 1)
|
||
|
||
for m in v9_diel_re.finditer(joined):
|
||
n = int(m.group(1))
|
||
diel_type = int(m.group(2))
|
||
height = 0.0
|
||
value = ""
|
||
hkey = f"V9_STACK_LAYER{n}_DIELHEIGHT="
|
||
mkey = f"V9_STACK_LAYER{n}_DIELMATERIAL="
|
||
for part in parts:
|
||
pu = part.upper()
|
||
if pu.startswith(hkey.upper()):
|
||
hraw = part.split("=", 1)[1]
|
||
hraw = re.sub(r"mil", "", hraw, flags=re.I).strip()
|
||
try:
|
||
height = float(hraw) * 0.0254
|
||
except ValueError:
|
||
height = 0.0
|
||
if pu.startswith(mkey.upper()):
|
||
value = part.split("=", 1)[1]
|
||
if value and height > 0:
|
||
materials.append(
|
||
DielProperty("DielMaterial", value, height, diel_type, n)
|
||
)
|
||
|
||
for m in v8_diel_re.finditer(joined):
|
||
n = int(m.group(1))
|
||
diel_type = int(m.group(2))
|
||
height = 0.0
|
||
value = ""
|
||
hkey = f"LAYER_V8_{n}_DIELHEIGHT="
|
||
mkey = f"LAYER_V8_{n}_DIELMATERIAL="
|
||
for part in parts:
|
||
pu = part.upper()
|
||
if pu.startswith(hkey.upper()):
|
||
hraw = part.split("=", 1)[1]
|
||
hraw = re.sub(r"mil", "", hraw, flags=re.I).strip()
|
||
try:
|
||
height = float(hraw) * 0.0254
|
||
except ValueError:
|
||
height = 0.0
|
||
if pu.startswith(mkey.upper()):
|
||
value = part.split("=", 1)[1]
|
||
if value and height > 0:
|
||
materials.append(
|
||
DielProperty("DielMaterial", value, height, diel_type, n)
|
||
)
|
||
|
||
if not found_layerset and layer_count > 0:
|
||
layer_count -= 1
|
||
|
||
data.pcb_layer_count = max(layer_count, 0)
|
||
data.pcb_diel_materials = materials
|
||
return True
|
||
|
||
|
||
def make_complex_string(text: str, props: list[ComponentProperty]) -> str:
|
||
from app.services.designators import resolve_expression
|
||
|
||
mapping = {p.name: p.text for p in props}
|
||
designator = mapping.get("Designator", "")
|
||
return sanitize_text(resolve_expression(text, mapping, designator))
|
||
|
||
|
||
def find_prjpcb(extract_dir: Path) -> Optional[Path]:
|
||
candidates = list(extract_dir.rglob("*.PrjPcb")) + list(extract_dir.rglob("*.prjpcb"))
|
||
return candidates[0] if candidates else None
|