diff --git a/web/backend/app/services/altium_parser.py b/web/backend/app/services/altium_parser.py index 2557944..27af848 100644 --- a/web/backend/app/services/altium_parser.py +++ b/web/backend/app/services/altium_parser.py @@ -57,20 +57,79 @@ class ProjectData: current_variant_number: int = 0 -def _read_cp1251_lines(path: Path) -> list[str]: - raw = path.read_bytes() - text = raw.decode("cp1251", errors="replace") +_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]: - """Read Altium binary-ish docs and extract pipe-separated ASCII chunks per line.""" + """SchDoc/PcbDoc are binary; desktop always reads as CP1251 text lines.""" 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()] +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 = "" @@ -85,7 +144,7 @@ class AltiumParser: data = ProjectData() data.prj_params_variant_list = [[]] # index 0 = No Variations - lines = _read_cp1251_lines(path) + lines = _read_text_lines(path) prev = "" prev_prev = "" for line in lines: @@ -98,7 +157,12 @@ class AltiumParser: 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: + 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 @@ -118,25 +182,64 @@ class AltiumParser: def _parse_project_variant_section(self, prj: str, data: ProjectData) -> None: upper = prj.upper() - if len(prj) >= 15 and upper.startswith("[PROJECTVARIANT"): + if len(prj) > 15 and upper.startswith("[PROJECTVARIANT"): + data.component_variant_prop_list.clear() 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) + 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 data.is_waiting_variant_description and upper.startswith("DESCRIPTION="): + 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) - data.is_waiting_variant_description = False + + 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 - if len(prj) >= 19 and upper.startswith("PARAMVARIATIONCOUNT"): - self._finalize_current_variant(data) + 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: - upper = prj.upper() + if len(prj) <= 9: + return parts = prj.split("|") if not parts: return @@ -144,35 +247,15 @@ class AltiumParser: 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) + 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 "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)) + 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 @@ -191,85 +274,91 @@ class AltiumParser: 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: + if len(prj) <= 13: 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(): + 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) < 19: + if len(prj) <= 13: 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) + 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_file(self, path: Path, components_list: list[list[ComponentProperty]]) -> None: - lines = _read_pipe_chunks(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(): - nonlocal component_props, is_no_bom, is_first_part, is_component + 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 line in lines: - parts = line.split("|") - i = 0 - while i < len(parts): - part = parts[i] - up = part.upper() + for i in range(len(parts) - 1): + 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 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 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 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