"""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 def _read_cp1251_lines(path: Path) -> list[str]: raw = path.read_bytes() text = raw.decode("cp1251", errors="replace") return [line.rstrip("\r\n") for line in text.splitlines()] def _read_pipe_chunks(path: Path) -> list[str]: """Read Altium binary-ish docs and extract pipe-separated ASCII chunks per line.""" raw = path.read_bytes() text = raw.decode("cp1251", errors="replace") # Also try latin-1 overlay for pipe records that may have been mangled return [line for line in text.splitlines() if line.strip()] 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_cp1251_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.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.is_waiting_variant_description = True # Extract number if present m = re.search(r"(\d+)", prj) data.current_variant_number = int(m.group(1)) if m else len(data.variant_names) while len(data.prj_params_variant_list) <= data.current_variant_number: data.prj_params_variant_list.append([]) return if data.is_waiting_variant_description and upper.startswith("DESCRIPTION="): name = prj[12:].strip() if name and name not in data.variant_names: data.variant_names.append(name) data.is_waiting_variant_description = False return if len(prj) >= 19 and upper.startswith("PARAMVARIATIONCOUNT"): self._finalize_current_variant(data) def _parse_component_variations(self, prj: str, data: ProjectData) -> None: upper = prj.upper() parts = prj.split("|") if not parts: return values = parts[0].split("=") if len(values) <= 2: return if "VARIATION" in values[0].upper() and "DESIGNATOR" in values[1].upper(): designator = values[2] prop_list = [ComponentProperty("Designator", designator)] kind = "0" if len(parts) > 2: kind_part = parts[2] if "=" in kind_part: kind = kind_part.split("=", 1)[1] prop_list.append(ComponentProperty("Kind", kind)) if kind == "1": data.dnf_designators_list.append(designator) else: data.fitted_designators_list.append(designator) # flush previous component props into list if data.component_variant_prop_list: # keep designator props as start of new component pass data.components_prop_variant_list.append(prop_list) return if "PARAMVARIATION" in values[0].upper() and "PARAMETERNAME" in values[1].upper(): prop_name = values[2] prop_text = "" if len(parts) > 1 and "=" in parts[1]: prop_text = parts[1].split("=", 1)[1] data.component_variant_prop_list.append(ComponentProperty(prop_name, prop_text)) # attach to last variation component if present if data.components_prop_variant_list: data.components_prop_variant_list[-1].append(ComponentProperty(prop_name, prop_text)) 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) < 19: return if prj[:13].upper() != "DOCUMENTPATH=": return if prj[-6:].upper() != "SCHDOC": return rel = prj[13:] sch_path = (prj_path.parent / rel).resolve() if sch_path.exists(): self._parse_schdoc_file(sch_path, data.components_list) def _parse_pcbdoc_files(self, prj: str, prj_path: Path, data: ProjectData) -> None: if len(prj) < 19: return if prj[:13].upper() != "DOCUMENTPATH=": return if prj[-6:].upper() != "PCBDOC": return rel = prj[13:] pcb_path = (prj_path.parent / rel).resolve() if pcb_path.exists(): if self._parse_pcbdoc_file(pcb_path, data): data.pcb_doc_file_name = str(pcb_path) def _parse_schdoc_file(self, path: Path, components_list: list[list[ComponentProperty]]) -> None: lines = _read_pipe_chunks(path) is_component = False is_no_bom = False is_first_part = True component_props: list[ComponentProperty] = [] def flush(): nonlocal component_props, is_no_bom, is_first_part, is_component 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 line in lines: parts = line.split("|") i = 0 while i < len(parts): part = parts[i] up = part.upper() if len(part) == 8 and up == "RECORD=1": if is_component: flush() is_component = True i += 1 continue if is_component: if up.startswith("COMPONENTKINDVERSION2=5"): is_no_bom = True elif up.startswith("CURRENTPARTID="): if up != "CURRENTPARTID=1" or len(part) > 15: is_first_part = False elif ( i + 1 < len(parts) and up.startswith("TEXT=") and parts[i + 1].upper().startswith("NAME=") ): text = part[5:] name = parts[i + 1][5:] if not is_no_bom: component_props.append(ComponentProperty(name, text)) i += 2 continue elif up.startswith("HEADER="): flush() is_component = False i += 1 if is_component: flush() 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: text = sanitize_text(text) if not text.startswith("="): return text expr = text[1:] mapping = {p.name: p.text for p in props} def repl(m: re.Match) -> str: key = m.group(1) return mapping.get(key, "") return sanitize_text(re.sub(r"['\"]([^'\"]+)['\"]", repl, expr)) 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