diff --git a/Caddyfile b/Caddyfile index 9483f83..bb746e5 100644 --- a/Caddyfile +++ b/Caddyfile @@ -1,9 +1,9 @@ {$DOMAIN} { - @comparison path /api/compare /api/v1/comparisons + @comparison path /api/compare /api/v1/comparisons /api/v1/pcbs/comparisons request_body @comparison { max_size 65MiB } - @single not path /api/compare /api/v1/comparisons + @single not path /api/compare /api/v1/comparisons /api/v1/pcbs/comparisons request_body @single { max_size 33MB } diff --git a/README.md b/README.md index 647ca84..978c604 100644 --- a/README.md +++ b/README.md @@ -148,4 +148,4 @@ The fixture downloader checksum-pins an actual upstream KiCad SchDoc. Unit-test ## Architecture and PCB extension -See [architecture](docs/ARCHITECTURE.md). `app/documents.py` is the parser registry, the schema preserves source records separately from vector primitives, `app/render.py` is the SVG serializer, and workers/storage are independent of schematic record decoding. A future PCB adapter can introduce layer-aware primitives and an appropriate viewer through these boundaries; no PCB parser or PCB rendering is shipped here. `.PcbDoc` uploads are rejected. +See [architecture](docs/ARCHITECTURE.md). `app/documents.py` is the parser registry, the schema preserves source records separately from vector primitives, `app/render.py` is the SVG serializer, and workers/storage are independent of schematic record decoding. Native PCB support uses separate `POST /api/v1/pcbs/render` and `POST /api/v1/pcbs/comparisons` endpoints, with layer-aware fullscreen viewers and structural comparison. Existing SchDoc endpoints remain unchanged and reject PCB uploads. See [PCB API, supported format subset and deployment](docs/PCB.md). diff --git a/THIRD_PARTY.md b/THIRD_PARTY.md index b10e805..0223f44 100644 --- a/THIRD_PARTY.md +++ b/THIRD_PARTY.md @@ -7,3 +7,5 @@ Runtime packages are installed with upstream package metadata/license files: Fas Format research is linked from `docs/COMPATIBILITY.md`. No GPL KiCad importer code is incorporated. The checksum-pinned KiCad QA fixture is downloaded to an ignored directory and excluded from Docker/source distribution; consult upstream repository and fixture licensing before redistributing it. The user's local designs are not included in project source or the image. Administrator-supplied fonts must have appropriate web-serving licenses. Embedded document images and symbols retain their owners' rights; uploading does not grant redistribution rights. + +The native PCB binary/geometry/parser modules also originate from that independently written local checkout and are adapted to this service. Format layouts were rechecked against the [KiCad developer format reference](https://dev-docs.kicad.org/en/import-formats/altium/index.html). No KiCad GPL importer implementation or proprietary PCB sample is incorporated. See [PCB provenance and compatibility](docs/PCB.md). diff --git a/backend/app/cleanup.py b/backend/app/cleanup.py index ef06ab6..670cd5d 100644 --- a/backend/app/cleanup.py +++ b/backend/app/cleanup.py @@ -4,6 +4,7 @@ import time from sqlalchemy import select, delete, or_ from .db import Job, sessions, engine from .storage import storage +from .job_types import comparison_sources log=logging.getLogger(__name__) @@ -15,8 +16,7 @@ def cleanup(batch=100): for job in db.scalars(q): try: storage().delete(job.input_key) - if job.input_key.endswith('/comparison.json'): - for side in ('old','new'): storage().delete(f'{job.id}/{side}.SchDoc') + for source in comparison_sources(job.input_key): storage().delete(source) for attempt in (1,2): for name in ('data.json','drawing.svg','old.svg','new.svg'): storage().delete(f'{job.id}/attempt-{attempt}/{name}') diff --git a/backend/app/documents.py b/backend/app/documents.py index 560f32f..b9e6ff5 100644 --- a/backend/app/documents.py +++ b/backend/app/documents.py @@ -2,12 +2,13 @@ from typing import Protocol from .models.schematic import Document from .parser.schdoc_parser import parse_schdoc +from .parser.pcb_parser import parse_pcbdoc from .parser.record_decoder import ParseError class DocumentParser(Protocol): def __call__(self, data: bytes, filename: str, identifier: str, **options) -> Document: ... -PARSERS: dict[str, DocumentParser] = {'schematic': parse_schdoc} +PARSERS: dict[str, DocumentParser] = {'schematic': parse_schdoc, 'pcb': parse_pcbdoc} def parse_document(data: bytes, filename: str, identifier: str, document_type='schematic', **options) -> Document: parser=PARSERS.get(document_type) diff --git a/backend/app/job_types.py b/backend/app/job_types.py new file mode 100644 index 0000000..ac99ca4 --- /dev/null +++ b/backend/app/job_types.py @@ -0,0 +1,20 @@ +"""Server-owned immutable input names classify jobs, including legacy uploads.""" +INPUT_KINDS = { + 'input.SchDoc': 'schematic_render', + 'comparison.json': 'schematic_comparison', + 'input.PcbDoc': 'pcb_render', + 'pcb-comparison.json': 'pcb_comparison', +} + + +def job_kind(input_key): + return INPUT_KINDS.get(input_key.rsplit('/', 1)[-1]) + + +def comparison_sources(input_key): + kind = job_kind(input_key) + if kind not in ('schematic_comparison', 'pcb_comparison'): + return () + prefix = input_key.rsplit('/', 1)[0] + suffix = 'PcbDoc' if kind == 'pcb_comparison' else 'SchDoc' + return tuple(f'{prefix}/{side}.{suffix}' for side in ('old', 'new')) diff --git a/backend/app/main.py b/backend/app/main.py index fcce0d5..3a33000 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -22,6 +22,7 @@ from .fonts import fonts, stylesheet from .models.api import UploadResponse, RenderStatus, ErrorEnvelope from .models.schematic import Document from .uploads import read_schematic +from .job_types import job_kind FRONTEND = Path(__file__).resolve().parents[2] / 'frontend' ID = re.compile(r'^[a-f0-9]{48}$') @@ -54,7 +55,7 @@ class UploadLimit: def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope['type'] != 'http': return await self.app(scope, receive, send) - file_count = 2 if scope['path'] in ('/api/compare','/api/v1/comparisons') else 1 + file_count = 2 if scope['path'] in ('/api/compare','/api/v1/comparisons','/api/v1/pcbs/comparisons') else 1 limit = file_count * settings().max_upload_bytes + 1024 * 1024 received = 0 async def bounded_receive(): @@ -71,7 +72,7 @@ app.add_middleware(UploadLimit) @app.middleware('http') async def security(request, call_next): # Reject unauthorized POST bodies before the multipart parser reads them. - if request.method == 'POST' and request.url.path in ('/api/v1/schematics/render','/api/compare','/api/v1/comparisons'): + if request.method == 'POST' and request.url.path in ('/api/v1/schematics/render','/api/compare','/api/v1/comparisons','/api/v1/pcbs/render','/api/v1/pcbs/comparisons'): if settings().api_key and not authenticated(request.headers.get('x-api-key')): return JSONResponse({'success': False, 'error': {'code': 'unauthorized', 'message': 'Provide a valid X-API-Key header'}}, 401) response = await call_next(request) @@ -109,10 +110,11 @@ def health(): with sessions()() as db: db.execute(text('SELECT 1')) return {'status': 'ok'} -def authorize(identifier, key, token, *, comparison=False): +def authorize(identifier, key, token, *, comparison=False, resource_kind=None): if not ID.fullmatch(identifier): error('not_found', 'Schematic not found', 404) with sessions()() as db: job = db.get(Job, identifier) - if job is None or job.input_key.endswith('/comparison.json') != comparison: + expected_kind = resource_kind or ('schematic_comparison' if comparison else 'schematic_render') + if job is None or job_kind(job.input_key) != expected_kind: error('not_found', 'Schematic not found', 404) # Private existence and expiry are not disclosed to unauthenticated callers. if job.access == 'private' and not (authenticated(key) or (token and hmac.compare_digest(job.token_hash, digest(token)))): @@ -201,7 +203,7 @@ def index(): return FileResponse(FRONTEND / 'index.html') @app.get('/assets/{name}', include_in_schema=False) def asset(name: str): - if name not in {'viewer.js','viewer.css','upload.js','upload.css','camera.js','compare.js','compare.css','compare-upload.js','interactions.js'}: error('not_found', 'Asset not found', 404) + if name not in {'viewer.js','viewer.css','upload.js','upload.css','camera.js','compare.js','compare.css','compare-upload.js','interactions.js','pcb.js','pcb.css','pcb-layers.js'}: error('not_found', 'Asset not found', 404) return FileResponse(FRONTEND / name) @app.get('/docs', include_in_schema=False) @@ -224,3 +226,5 @@ async def database_error(request, exc): # Comparison jobs share the existing queue, access policy and error envelope. from .compare_api import register_comparison_routes register_comparison_routes(app, authorize=authorize, require_api=require_api, error=error) +from .pcb_api import register_pcb_routes +register_pcb_routes(app, authorize=authorize, require_api=require_api, error=error) diff --git a/backend/app/models/pcb.py b/backend/app/models/pcb.py new file mode 100644 index 0000000..7c4c346 --- /dev/null +++ b/backend/app/models/pcb.py @@ -0,0 +1,67 @@ +"""PCB contracts extend geometry without changing the schematic wire schema.""" +from typing import Literal +from pydantic import BaseModel, Field +from .schematic import Document, Primitive, SchematicObject, Point, Bounds +from .api import RenderStatus + + +class PCBPrimitive(Primitive): + layer_ids: list[str] = Field(default_factory=list) + + +class PCBObject(SchematicObject): + native_uid: str = '' + layer_ids: list[str] = Field(default_factory=list) + + +class Footprint(BaseModel): + id: str + designator: str = '' + name: str = '' + value: str = '' + comment: str = '' + description: str = '' + library_reference: str = '' + source_library: str = '' + position: Point + rotation: float = 0 + mirrored: bool = False + bounds: Bounds + parameters: dict[str, str] = Field(default_factory=dict) + primitive_ids: list[str] = Field(default_factory=list) + object_ids: list[str] = Field(default_factory=list) + + +class PCBData(Document): + document_type: Literal['pcb'] = 'pcb' + components: list[Footprint] + primitives: list[PCBPrimitive] + objects: list[PCBObject] + + +class PCBStatus(RenderStatus): + document_type: Literal['pcb', 'pcb_comparison'] + status_url: str + layers: list[dict] | None = None + net_count: int | None = None + + +class PCBUploadResponse(PCBStatus): + viewer_token: str | None = None + + +class PCBComparisonStatus(PCBStatus): + comparison_id: str + url: str + summary: dict[Literal['added','removed','modified','moved'], int] | None = None + total_changed: int | None = None + changes: list[dict] | None = None + old: dict | None = None + new: dict | None = None + nets: dict | None = None + matching: dict | None = None + limitations: list[str] | None = None + + +class PCBComparisonUploadResponse(PCBComparisonStatus): + viewer_token: str | None = None diff --git a/backend/app/parser/pcb_binary.py b/backend/app/parser/pcb_binary.py new file mode 100644 index 0000000..762d195 --- /dev/null +++ b/backend/app/parser/pcb_binary.py @@ -0,0 +1,106 @@ +"""Checked Altium PCB framing. No embedded content is decompressed or executed.""" + +import math +import struct +from .record_decoder import ParseError, properties + +MAX_ITEMS = 100_000 +MAX_BLOCK = 16 * 1024 * 1024 + + +class Reader: + def __init__(self, data, label="PCB"): + self.data = memoryview(data) + self.offset = 0 + self.label = label + + @property + def remaining(self): + return len(self.data) - self.offset + + def take(self, length): + if length < 0 or length > self.remaining: + raise ParseError( + f"{self.label}: truncated data at byte {self.offset}, need {length}" + ) + start = self.offset + self.offset += length + return self.data[start : self.offset] + + def read(self, fmt): + result = struct.unpack("<" + fmt, self.take(struct.calcsize("<" + fmt))) + if any(isinstance(v, float) and not math.isfinite(v) for v in result): + raise ParseError(f"{self.label}: non-finite binary number") + return result[0] if len(result) == 1 else result + + def block(self): + length = self.read("I") + if length > MAX_BLOCK: + raise ParseError(f"{self.label}: subrecord exceeds {MAX_BLOCK} bytes") + return self.take(length) + + def pascal(self): + return bytes(self.take(self.read("B"))).decode("cp1252", errors="replace") + + +def text_records(data, name): + reader = Reader(data, name) + result = [] + while reader.remaining: + if len(result) >= MAX_ITEMS: + raise ParseError(f"{name}: excessive record count") + result.append(properties(bytes(reader.block()))) + return result + + +def binary_records(data, name, tag, blocks=1): + reader = Reader(data, name) + index = 0 + while reader.remaining: + if index >= MAX_ITEMS: + raise ParseError(f"{name}: excessive record count") + actual = reader.read("B") + if actual != tag: + raise ParseError( + f"{name}: unsupported tag {actual} at record {index}, expected {tag}" + ) + yield index, [bytes(reader.block()) for _ in range(blocks)] + index += 1 + + +def wide_strings(data): + reader = Reader(data, "WideStrings6") + result = {} + while reader.remaining: + if len(result) >= MAX_ITEMS: + raise ParseError("Excessive wide string count") + index = reader.read("I") + length = reader.read("I") + # Altium writes empty strings with length=2 but no payload in some exports. + if ( + length == 2 + and bytes(reader.data[reader.offset : reader.offset + 2]) != b"\0\0" + ): + payload = b"" + else: + if length > MAX_BLOCK: + raise ParseError("Wide string exceeds size limit") + payload = reader.take(length) + if len(payload) % 2: + raise ParseError("Odd UTF-16 string length") + result[index] = ( + bytes(payload).decode("utf-16le", errors="replace").rstrip("\x00") + ) + return result + + +def at(data, offset, fmt, default=None): + size = struct.calcsize("<" + fmt) + if offset < 0 or offset + size > len(data): + if default is not None: + return default + raise ParseError(f"Truncated PCB field at offset {offset}, need {size} bytes") + values = struct.unpack_from("<" + fmt, data, offset) + if any(isinstance(v, float) and not math.isfinite(v) for v in values): + raise ParseError("Non-finite PCB field") + return values[0] if len(values) == 1 else values diff --git a/backend/app/parser/pcb_container.py b/backend/app/parser/pcb_container.py new file mode 100644 index 0000000..35c9552 --- /dev/null +++ b/backend/app/parser/pcb_container.py @@ -0,0 +1,44 @@ +"""Cheap bounded PCB envelope validation; complete decoding stays in the worker.""" +import io +import struct +import olefile +from .record_decoder import ParseError +from .pcb_binary import Reader, text_records + +SIGNATURE = b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1' + + +def identify_pcb(data): + if not data.startswith(SIGNATURE): + raise ParseError('PcbDoc must be an OLE compound document') + try: + with olefile.OleFileIO(io.BytesIO(data), raise_defects=olefile.DEFECT_INCORRECT) as ole: + if len(ole.listdir()) > 1024: + raise ParseError('Excessive PCB stream count') + for name, limit in [('FileHeaderSix', 4096), ('Board6/Data', 4 * 1024**2)]: + if not ole.exists(name) or ole.get_type(name) != 2: + raise ParseError(f'Missing PCB stream {name}') + if ole.get_size(name) > limit: + raise ParseError(f'PCB stream {name} exceeds size limit') + reader = Reader(ole.openstream('FileHeaderSix').read(), 'FileHeaderSix') + length = reader.read('I'); envelope = reader.pascal() + if length != len(envelope) or envelope != 'PCB 6.0 Binary File': + raise ParseError('Unsupported PCB file envelope') + records = text_records(ole.openstream('Board6/Data').read(), 'Board6') + if len(records) != 1 or records[0].get('KIND') != 'Protel_Advanced_PCB': + raise ParseError('Invalid Board6 identification') + if records[0].get('VERSION') not in ('5.0', '5.01', '6.0'): + raise ParseError('Unsupported PCB board version') + except (OSError, IOError, ValueError, IndexError, struct.error) as exc: + if isinstance(exc, ParseError): raise + raise ParseError('Invalid PCB compound document') from exc + + +def contains_pcb(data): + """Reject PCB data renamed to SchDoc without altering legacy corrupt-file handling.""" + if not data.startswith(SIGNATURE): return False + try: + with olefile.OleFileIO(io.BytesIO(data)) as ole: + return ole.exists('Board6/Data') + except (OSError, ValueError, IndexError, struct.error): + return False diff --git a/backend/app/parser/pcb_geometry.py b/backend/app/parser/pcb_geometry.py new file mode 100644 index 0000000..4f2052f --- /dev/null +++ b/backend/app/parser/pcb_geometry.py @@ -0,0 +1,155 @@ +import math +import re +from app.models.schematic import Point, Bounds +from .geometry import bounds, poly_path, transform_point +from .record_decoder import ParseError + +# PCB integer coordinate unit = 1/10000 mil. Common viewer unit = 10 mil. +BINARY_SCALE = 100000 + + +def raw_unit(value): + n = value / BINARY_SCALE + if not math.isfinite(n) or abs(n) > 100000: + raise ParseError("Invalid or out-of-range PCB coordinate") + return n + + +def unit(value): + match = re.fullmatch(r"\s*([-+\d.eE]+)\s*(mil|mm|in|inch)?\s*", str(value), re.I) + if not match: + raise ParseError(f"Invalid PCB unit: {str(value)[:80]}") + n = float(match[1]) + suffix = (match[2] or "mil").lower() + factor = {"mil": 0.1, "mm": 1 / 0.254, "in": 100, "inch": 100}[suffix] + result = n * factor + if not math.isfinite(result) or abs(result) > 100000: + raise ParseError("PCB unit out of bounds") + return result + + +def xy(x, y): + return Point(x=raw_unit(x), y=-raw_unit(y)) + + +def ellipse_path(cx, cy, rx, ry=None): + ry = rx if ry is None else ry + return f"M{cx - rx:.15g},{cy:.15g} A{rx:.15g},{ry:.15g} 0 1 0 {cx + rx:.15g},{cy:.15g} A{rx:.15g},{ry:.15g} 0 1 0 {cx - rx:.15g},{cy:.15g} Z" + + +def arc_path(center, radius, start, end): + if not all(math.isfinite(v) for v in (radius,start,end)): raise ParseError("Non-finite arc geometry") + if radius <= 0: + raise ParseError("Arc radius must be positive") + if abs(end - start) >= 359.99999 or start == end: + return ellipse_path(center.x, center.y, radius) + a, b = math.radians(start), math.radians(end) + x1, y1 = center.x + radius * math.cos(a), center.y - radius * math.sin(a) + x2, y2 = center.x + radius * math.cos(b), center.y - radius * math.sin(b) + return f"M{x1:.15g},{y1:.15g} A{radius:.15g},{radius:.15g} 0 {int((end - start) % 360 > 180)} 0 {x2:.15g},{y2:.15g}" + + +def pad_shape(position, w, h, shape, rotation, radius_percent=0): + if w <= 0 or h <= 0: + raise ParseError("Pad dimensions must be positive") + # Build in local coordinates, then place absolute points. Exact arcs use a path transform. + angle = math.radians(-rotation) + + def q(x, y): + return Point( + x=position.x + x * math.cos(angle) - y * math.sin(angle), + y=position.y + x * math.sin(angle) + y * math.cos(angle), + ) + + def pair(p): + return f"{p.x:.15g},{p.y:.15g}" + + if ( + shape == 1 + ): # Circle or oval/obround; unequal sizes form a capsule, not an ellipse. + r = min(w, h) / 2 + elif shape == 9: + r = min(w, h) * radius_percent / 200 + else: + r = 0 + if shape == 3: + # Altium octagon chamfers at 45 degrees. + c = min(w, h) / (2 + math.sqrt(2)) + pts = [ + q(x, y) + for x, y in [ + (-w / 2 + c, -h / 2), + (w / 2 - c, -h / 2), + (w / 2, -h / 2 + c), + (w / 2, h / 2 - c), + (w / 2 - c, h / 2), + (-w / 2 + c, h / 2), + (-w / 2, h / 2 - c), + (-w / 2, -h / 2 + c), + ] + ] + return poly_path(pts, True), bounds(pts) + if shape not in (1, 2, 9): + raise ParseError(f"Unsupported pad shape {shape}") + corners = [ + q(x, y) + for x, y in [(-w / 2, -h / 2), (w / 2, -h / 2), (w / 2, h / 2), (-w / 2, h / 2)] + ] + if r <= 0: + return poly_path(corners, True), bounds(corners) + pts = [ + q(x, y) + for x, y in [ + (-w / 2 + r, -h / 2), + (w / 2 - r, -h / 2), + (w / 2, -h / 2 + r), + (w / 2, h / 2 - r), + (w / 2 - r, h / 2), + (-w / 2 + r, h / 2), + (-w / 2, h / 2 - r), + (-w / 2, -h / 2 + r), + ] + ] + d = f"M{pair(pts[0])} L{pair(pts[1])} A{r:.15g},{r:.15g} {-rotation:.15g} 0 1 {pair(pts[2])} L{pair(pts[3])} A{r:.15g},{r:.15g} 0 0 1 {pair(pts[4])} L{pair(pts[5])} A{r:.15g},{r:.15g} 0 0 1 {pair(pts[6])} L{pair(pts[7])} A{r:.15g},{r:.15g} 0 0 1 {pair(pts[0])} Z" + return d, bounds(corners) + + +def property_outline(p): + vertices = [] + for i in range(20000): + if f"VX{i}" not in p: + break + if f"VY{i}" not in p: raise ParseError("PCB outline vertex is missing Y") + if int(p.get(f"KIND{i}",0)) and not all(f"{key}{i}" in p for key in ("CX","CY","R","SA","EA")): + raise ParseError("PCB outline arc is missing native geometry") + vertices.append( + ( + Point(x=unit(p[f"VX{i}"]), y=-unit(p.get(f"VY{i}", "0mil"))), + int(p.get(f"KIND{i}", 0)), + Point( + x=unit(p.get(f"CX{i}", "0mil")), y=-unit(p.get(f"CY{i}", "0mil")) + ), + unit(p.get(f"R{i}", "0mil")), + float(p.get(f"SA{i}", 0)), + float(p.get(f"EA{i}", 0)), + ) + ) + if not vertices: + return "", bounds([]) + pts = [v[0] for v in vertices] + commands = [f"M{pts[0].x:.15g},{pts[0].y:.15g}"] + for i, (pos, kind, center, r, sa, ea) in enumerate(vertices): + if kind and r > 0: + # Arc belongs to its starting vertex, ending at the next vertex. + arc = arc_path(center, r, sa, ea) + commands.append("L" + arc.split(" ", 1)[0][1:]) + commands.append(arc.split(" ", 1)[1]) + pts.extend( + [ + Point(x=center.x - r, y=center.y - r), + Point(x=center.x + r, y=center.y + r), + ] + ) + nxt = vertices[(i + 1) % len(vertices)][0] + commands.append(f"L{nxt.x:.15g},{nxt.y:.15g}") + return " ".join(commands) + " Z", bounds(pts) diff --git a/backend/app/parser/pcb_parser.py b/backend/app/parser/pcb_parser.py new file mode 100644 index 0000000..069fa87 --- /dev/null +++ b/backend/app/parser/pcb_parser.py @@ -0,0 +1,947 @@ +"""Independent, bounded reader for the documented PcbDoc 6.0 binary record family.""" + +import hashlib +import io +import math +import re +import struct +from collections import Counter +import olefile +from app.models.schematic import Sheet, Point, Bounds, Diagnostic, BoardLayer, Net +from app.models.pcb import PCBData as Document, Footprint as Component, PCBObject as SchematicObject, PCBPrimitive as Primitive +from .record_decoder import ParseError, properties +from .pcb_binary import Reader, at, text_records, binary_records, wide_strings +from .pcb_geometry import ( + raw_unit, + unit, + xy, + ellipse_path, + arc_path, + pad_shape, + property_outline, +) +from .geometry import bounds, union, poly_path, transform_point + +MAX_TOTAL = 64 * 1024 * 1024 +SPECIAL = { + 32: ("Bottom Layer", "#5484ed", "copper"), + 33: ("Top Overlay", "#f6e4b1", "silkscreen"), + 34: ("Bottom Overlay", "#c29ef1", "silkscreen"), + 35: ("Top Paste", "#92b8a0", "paste"), + 36: ("Bottom Paste", "#81a0b5", "paste"), + 37: ("Top Solder", "#966ac1", "mask"), + 38: ("Bottom Solder", "#b78b99", "mask"), + 55: ("Drill Guide", "#d5d8dc", "mechanical"), + 56: ("Keep-Out Layer", "#d374dc", "mechanical"), + 73: ("Drill Drawing", "#cccbd0", "mechanical"), + 74: ("Multi-Layer", "#d9b660", "copper"), + 81: ("Drill holes", "#08151b", "drill"), + 0: ("Board Outline", "#67bc9d", "outline"), +} + + +def layer_number(name): + name = name.upper().replace(" ", "").replace("_", "") + fixed = { + "TOP": 1, + "TOPLAYER": 1, + "BOTTOM": 32, + "BOTTOMLAYER": 32, + "TOPOVERLAY": 33, + "BOTTOMOVERLAY": 34, + "TOPPASTE": 35, + "BOTTOMPASTE": 36, + "TOPSOLDER": 37, + "BOTTOMSOLDER": 38, + "KEEPOUT": 56, + "KEEPOUTLAYER": 56, + "MULTILAYER": 74, + "DRILLGUIDE": 55, + "DRILLDRAWING": 73, + } + if name in fixed: + return fixed[name] + m = re.fullmatch(r"(MID(?:LAYER)?|PLANE|MECHANICAL)(\d+)", name) + if m: + n = int(m[2]) + if m[1].startswith("MID"): + return n + 1 + if m[1] == "PLANE": + return n + 38 + return n + 56 if n <= 16 else 0x01020000 + n + return 0 + + +def resolve_layer(legacy, extended=0): + # The standard copper/overlay IDs remain authoritative in V6 records. + if 0x01020011 <= extended <= 0x0102FFFF: + return extended + return legacy + + +def parse_pcbdoc(data: bytes, filename: str, document_id: str, **options) -> Document: + if not data.startswith(b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"): + raise ParseError("PcbDoc must be an OLE compound document") + try: + ole = olefile.OleFileIO( + io.BytesIO(data), raise_defects=olefile.DEFECT_INCORRECT + ) + except (OSError, ValueError, struct.error) as exc: + raise ParseError(f"Invalid PCB container: {exc}") from exc + with ole: + entries = ole.listdir(streams=True, storages=False) + if len(entries) > 1024: + raise ParseError("Excessive PCB stream count") + streams = [{"name": "/".join(e), "size": ole.get_size(e)} for e in entries] + if ( + any(s["size"] > 32 * 1024 * 1024 for s in streams) + or sum(s["size"] for s in streams) > MAX_TOTAL + ): + raise ParseError("PCB stream size limit exceeded") + + def read(name, required=False): + if not ole.exists(name): + if required: + raise ParseError(f"Missing PCB stream {name}") + return b"" + return ole.openstream(name).read() + + header = read("FileHeaderSix", True) + hreader = Reader(header, "FileHeaderSix") + header_length = hreader.read("I") + envelope = hreader.pascal() + if header_length != len(envelope): + raise ParseError("PCB version string length mismatch") + if envelope != "PCB 6.0 Binary File": + raise ParseError(f"Unsupported PCB envelope: {envelope[:100]}") + board_records = text_records(read("Board6/Data", True), "Board6") + if len(board_records) != 1: + raise ParseError("Expected exactly one Board6 property record") + board = board_records[0] + if board.get("KIND") != "Protel_Advanced_PCB" or board.get("VERSION") not in ( + "5.0", + "5.01", + "6.0", + ): + raise ParseError(f"Unsupported PCB board version {board.get('VERSION')}") + diagnostics = [] + grouped = {} + counts = {} + variants = {} + objects = [] + primitives = [] + layers = {} + component_boxes = {} + + def report(code, message, record=None, severity="warning"): + key = (code, message) + if key in grouped: + grouped[key].count += 1 + return + d = Diagnostic( + severity=severity, code=code, message=message, record_id=record + ) + diagnostics.append(d) + grouped[key] = d + + def ensure_layer(n): + key = str(n) + if key in layers: + return layers[key] + if n == 1: + name, col, category = "Top Layer", "#d5624e", "copper" + elif 2 <= n <= 31: + name, col, category = ( + f"Mid Layer {n - 1}", + ("#a5bd59", "#b67ac7", "#d19e4d", "#63bdb5")[(n - 2) % 4], + "copper", + ) + elif n in SPECIAL: + name, col, category = SPECIAL[n] + elif 39 <= n <= 54: + name, col, category = f"Internal Plane {n - 38}", "#698a68", "plane" + elif 57 <= n <= 72 or n >= 0x01020000: + name, col, category = ( + f"Mechanical {n - 56 if n < 100 else n - 0x01020000}", + "#8196a8", + "mechanical", + ) + else: + name, col, category = f"Layer {n}", "#8196a8", "unknown" + name = board.get(f"LAYER{n}NAME", name) + if n >= 0x01020000: + for k, v in board.items(): + if ( + k.startswith("LAYERV7_") + and k.endswith("LAYERID") + and v == str(n) + ): + name = board.get(k[:-7] + "NAME", name) + order = ( + 0 + if n == 0 + else 1000 + if n == 81 + else 900 + if n == 33 + else 890 + if n == 34 + else 800 + if n == 74 + else 700 + if n == 1 + else 600 + if n == 32 + else 500 - n + if 2 <= n <= 31 + else 850 + ) + layer = BoardLayer( + id=key, + name=name, + color=col, + category=category, + visible=n in (0, 1, 33, 74, 81), + order=order, + ) + layers[key] = layer + return layer + + def props_records(name): + result = text_records(read(name + "/Data"), name) + check_count(name, len(result)) + return result + + def check_count(name, count): + counts[name] = count + hdr = read(name + "/Header") + if len(hdr) == 4 and at(hdr, 0, "I") != count: + report( + "record_count_mismatch", + f"{name}: header declares {at(hdr, 0, 'I')}, decoded {count}", + ) + + def typed(name, tag, blocks=1): + count = 0 + for idx, parts in binary_records(read(name + "/Data"), name, tag, blocks): + count += 1 + signature = ",".join(str(len(p)) for p in parts) + variants.setdefault(name, Counter())[signature] += 1 + if sum(counts.values()) + count > 100000: + raise ParseError("Excessive PCB object count") + yield idx, parts + check_count(name, count) + + uid_map = {} + uid_stream = text_records(read("UniqueIDPrimitiveInformation/Data"), "UniqueIDPrimitiveInformation") + for item in uid_stream: + key = (item.get("PRIMITIVEOBJECTID", "").lower(), int(item.get("PRIMITIVEINDEX", -1))) + if key in uid_map: raise ParseError("Duplicate primitive UID mapping") + uid_map[key] = item.get("UNIQUEID", "") + + nets = [ + Net(id=f"net:{i}", name=p.get("NAME", f"Net {i}"), properties=p) + for i, p in enumerate(props_records("Nets6")) + ] + components = [] + for i, p in enumerate(props_records("Components6")): + if not {"X","Y","LAYER"}.issubset(p): + raise ParseError(f"Components6:{i}: missing native placement fields") + pos = Point(x=unit(p.get("X", "0mil")), y=-unit(p.get("Y", "0mil"))) + components.append( + Component( + id=f"Components6:{i}", + designator=p.get("SOURCEDESIGNATOR", ""), + name=p.get("SOURCELIBREFERENCE", p.get("PATTERN", "")), + value=p.get("COMMENT", ""), + comment=p.get("COMMENT", ""), + description=p.get("SOURCEDESCRIPTION", ""), + library_reference=p.get("PATTERN", ""), + source_library=p.get("SOURCEFOOTPRINTLIBRARY", ""), + position=pos, + rotation=float(p.get("ROTATION", 0)), + mirrored=layer_number(p.get("LAYER", "TOP")) == 32, + bounds=bounds([pos], 1), + parameters=p, + ) + ) + + if any(not math.isfinite(c.rotation) for c in components): + raise ParseError("Non-finite footprint rotation") + + def common(name, index, p, ext=0): + if len(p) < 13: + raise ParseError(f"{name}:{index}: short primitive header") + net, component = at(p, 3, "H"), at(p, 7, "H") + if net != 65535 and net >= len(nets): + raise ParseError(f"{name}:{index}: invalid net index {net}") + if component != 65535 and component >= len(components): + raise ParseError(f"{name}:{index}: invalid component index {component}") + lid = resolve_layer(p[0], ext) + layer = ensure_layer(lid) + c = components[component] if component != 65535 else None + props = { + "LAYER_ID": str(lid), + "NET_INDEX": str(net), + "NET_NAME": nets[net].name if net != 65535 else "", + "COMPONENT_INDEX": str(component), + "LOCKED": str(not bool(p[1] & 4)), + "FLAGS_1": str(p[1]), + "FLAGS_2": str(p[2]), + } + obj = SchematicObject( + id=f"{name}:{index}", + kind=name.removesuffix("6").lower(), + native_uid=uid_map.get(({"Tracks6":"track","Arcs6":"arc","Pads6":"pad","Vias6":"via","Texts6":"text","Fills6":"fill","Regions6":"region","ShapeBasedRegions6":"region"}.get(name,""),index), ""), + record_type={ + "Tracks6": 4, + "Arcs6": 1, + "Pads6": 2, + "Vias6": 3, + "Texts6": 5, + "Fills6": 6, + "Regions6": 11, + "ShapeBasedRegions6": 11, + }.get(name, 0), + source_stream=name, + source_index=index, + owner_id=c.id if c else None, + component_id=c.id if c else None, + layer_id=layer.id, + net_id=nets[net].id if net != 65535 else None, + position=Point(x=0, y=0), + bounds=bounds([]), + properties=props, + ) + objects.append(obj) + layer.object_count += 1 + if c: + c.object_ids.append(obj.id) + return obj, c + + def shape( + obj, + path, + box, + fill="none", + width=0.0, + layer_id=None, + role="", + stroke=None, + fill_rule="nonzero", + position=None, + text="", + rotation=0, + mirrored=False, + font_size=10, + font_family="Arial", + bold=False, + italic=False, + ): + layer = ensure_layer( + int(layer_id if layer_id is not None else obj.layer_id or 0) + ) + col = layer.color + primitive = Primitive( + id=f"{obj.id}:p{len(obj.primitive_ids)}", + kind="text" if text else "path", + object_id=obj.id, + component_id=obj.component_id, + layer_id=layer.id, + net_id=obj.net_id, + role=role, + path=path, + bounds=box, + stroke=col if stroke is None else stroke, + fill=col if fill == "layer" else fill, + width=width, + fill_rule=fill_rule, + position=position or obj.position, + text=text, + rotation=rotation, + mirrored=mirrored, + font_size=font_size, + font_family=font_family, + bold=bold, + italic=italic, + ) + if len(primitives) >= 250000: raise ParseError("PCB drawable limit exceeded") + primitives.append(primitive) + obj.primitive_ids.append(primitive.id) + obj.bounds = union([obj.bounds, box]) if len(obj.primitive_ids) > 1 else box + if obj.component_id: + ci = int(obj.component_id.split(":")[1]) + components[ci].primitive_ids.append(primitive.id) + if layer.category not in ("mechanical", "paste", "mask"): + component_boxes.setdefault(obj.component_id, []).append(box) + return primitive + + # Outline is extracted from actual Board6 vertices, including rounded corners. + outline, board_bounds = property_outline(board) + if not outline: + raise ParseError("PCB has no supported Board6 outline") + outline_obj = SchematicObject( + id="Board6:0", + kind="board_outline", + record_type=0, + source_stream="Board6", + source_index=0, + layer_id="0", + position=Point(x=board_bounds.x, y=board_bounds.y), + bounds=board_bounds, + properties=board, + ) + objects.append(outline_obj) + shape( + outline_obj, outline, board_bounds, fill="#13312e", width=0.18, role="board" + ) + counts["Board6"] = 1 + # Preserve polygon definitions as outlines only; the filled copper is read from Regions6. + polygons = props_records("Polygons6") + for i, p in enumerate(polygons): + lid = layer_number(p.get("LAYER", "TOP")) + layer = ensure_layer(lid) + path, box = property_outline(p) + net = int(p.get("NET", 65535)) + if net != 65535 and not 0 <= net < len(nets): raise ParseError("Invalid polygon net reference") + o = SchematicObject( + id=f"Polygons6:{i}", + kind="polygon_definition", + native_uid=p.get("UNIQUEID", ""), + record_type=0, + source_stream="Polygons6", + source_index=i, + layer_id=layer.id, + net_id=nets[net].id if 0 <= net < len(nets) else None, + visible=False, + position=Point(x=box.x, y=box.y), + bounds=box, + properties=p, + ) + objects.append(o) + for i, (p,) in typed("Tracks6", 4): + obj, c = common("Tracks6", i, p, at(p, 41, "I", 0)) + a, b = xy(*at(p, 13, "ii")), xy(*at(p, 21, "ii")) + w = raw_unit(at(p, 29, "i")) + if w < 0: + raise ParseError("Negative track width") + obj.position = a + obj.connection_points = [a, b] + obj.kind = "track" + obj.properties["WIDTH_MIL"] = str(w * 10) + polygon = at(p, 5, "H") + obj.properties["POLYGON_INDEX"] = str(polygon) + shape( + obj, + poly_path([a, b]), + bounds([a, b], w / 2), + width=w, + role="polygon_outline" if p[1] & 2 else "track", + ) + for i, (p,) in typed("Arcs6", 1): + obj, c = common("Arcs6", i, p, at(p, 52, "I", 0)) + pos = xy(*at(p, 13, "ii")) + r = raw_unit(at(p, 21, "i")) + start, end = at(p, 25, "dd") + w = raw_unit(at(p, 41, "i")) + if w < 0: raise ParseError("Negative arc width") + obj.kind = "arc" + obj.position = pos + obj.properties.update( + { + "RADIUS_MIL": str(r * 10), + "START_ANGLE": str(start), + "END_ANGLE": str(end), + "WIDTH_MIL": str(w * 10), + } + ) + shape( + obj, + arc_path(pos, r, start, end), + Bounds( + x=pos.x - r - w / 2, + y=pos.y - r - w / 2, + width=2 * r + w, + height=2 * r + w, + ), + width=w, + role="track", + ) + for i, (p,) in typed("Fills6", 6): + obj, c = common("Fills6", i, p, at(p, 42, "I", 0)) + a, b = xy(*at(p, 13, "ii")), xy(*at(p, 21, "ii")) + rotation = at(p, 29, "d") + center = Point(x=(a.x + b.x) / 2, y=(a.y + b.y) / 2) + obj.position = center + obj.kind = "fill" + path, box = pad_shape(center, abs(b.x - a.x), abs(b.y - a.y), 2, rotation) + obj.properties["ROTATION"] = str(rotation) + shape(obj, path, box, fill="layer", role="fill") + + # Render saved tessellated regions with exact interior holes. Keep definition streams separate. + def regions(name, extended): + for i, (p,) in typed(name, 11): + obj, c = common(name, i, p) + obj.kind = "region" + reader = Reader(p, name) + reader.take(14) + hole_count = reader.read("H") + reader.take(2) + props = properties(bytes(reader.block())) + obj.properties.update(props) + n = reader.read("I") + (1 if extended else 0) + if n > 50000 or hole_count > 20000: + raise ParseError("Excessive region geometry") + vertices = [] + commands = [] + all_points = [] + for _ in range(n): + if extended: + curved = reader.read("B") + pos = xy(*reader.read("ii")) + center = xy(*reader.read("ii")) + r = raw_unit(reader.read("i")) + sa, ea = reader.read("dd") + vertices.append((pos, curved, center, r, sa, ea)) + else: + vertices.append((xy(*reader.read("dd")), 0, None, 0, 0, 0)) + for j, (pos, curved, center, r, sa, ea) in enumerate(vertices): + commands.append(("M" if j == 0 else "L") + f"{pos.x:.15g},{pos.y:.15g}") + all_points.append(pos) + if curved and r > 0: + arc = arc_path(center, r, sa, ea) + if " " in arc: + commands.extend( + ["L" + arc.split(" ", 1)[0][1:], arc.split(" ", 1)[1]] + ) + all_points.extend( + [ + Point(x=center.x - r, y=center.y - r), + Point(x=center.x + r, y=center.y + r), + ] + ) + commands.append("Z") + for _ in range(hole_count): + count = reader.read("I") + if count > 50000: + raise ParseError("Excessive region hole points") + pts = [xy(*reader.read("dd")) for _ in range(count)] + commands.append(poly_path(pts, True)) + all_points.extend(pts) + obj.properties["OUTLINE_PATH"] = " ".join(commands) + obj.bounds = bounds(all_points) + obj.position = all_points[0] if all_points else Point(x=0, y=0) + obj.properties["HOLE_COUNT"] = str(hole_count) + obj.properties["VERTEX_COUNT"] = str(n) + obj.properties["POLYGON_INDEX"] = str(at(p, 5, "H")) + kind = int(props.get("KIND", 0)) + polygon = at(p, 5, "H") + is_cutout = props.get("ISBOARDCUTOUT", "FALSE").upper() == "TRUE" + if extended: + # When saved fills exist, shape definitions must not cover clearances twice. + if polygon != 65535 and not is_cutout: + obj.visible = False + continue + # Standalone shapes are taken from this stream, with circular arcs intact. + role = "region" + else: + if polygon == 65535 and not is_cutout: + obj.visible = False + continue + role = "pour" + if kind != 0 and not is_cutout: + obj.visible = False + report( + "region_definition", + f"Region kind {kind} is retained as a definition; saved copper holes govern the fill", + obj.id, + "info", + ) + continue + if is_cutout: + shape( + obj, + " ".join(commands), + obj.bounds, + fill="#08151b", + layer_id="0", + role="board_cutout", + fill_rule="evenodd", + ) + elif len(vertices) >= 3: + shape( + obj, + " ".join(commands), + obj.bounds, + fill="layer", + role=role, + fill_rule="evenodd", + ) + if reader.remaining: + report( + "region_extension", + f"{name} contains uninterpreted region extension bytes", + obj.id, + ) + + regions("Regions6", False) + regions("ShapeBasedRegions6", True) + poured = {o.properties.get("POLYGON_INDEX") for o in objects if o.source_stream == "Regions6" and o.visible} + for index, definition in enumerate(polygons): + if str(index) not in poured: + report("polygon_without_saved_pour", "Polygons6 definition has no decoded saved copper pour; no fill is invented", f"Polygons6:{index}") + # Active copper layers are obtained from legacy stack links, not from LAYERSETSCOUNT (view presets). + copper_ids = [1] + while True: + nxt = int(board.get(f"LAYER{copper_ids[-1]}NEXT", "0")) + if not nxt: break + if nxt in copper_ids or nxt not in range(1,55) or 33 <= nxt <= 38: + raise ParseError("Invalid copper stack links") + copper_ids.append(nxt) + if 32 not in copper_ids: + copper_ids = [1,32] + report("copper_stack", "No complete layer chain; only explicit top/bottom copper stack supported") + for index, lid in enumerate(copper_ids): + ensure_layer(lid).order = 700 - index * 10 + for i, parts in typed("Pads6", 2, 6): + name = Reader(parts[0], "Pad name").pascal() + p, stack = parts[4], parts[5] + if len(p) < 110: + raise ParseError("Unsupported short pad geometry") + obj, c = common("Pads6", i, p) + obj.kind = "pad" + pos = xy(*at(p, 13, "ii")) + obj.position = pos + sizes = [ + (raw_unit(x), raw_unit(y)) + for x, y in [at(p, 21, "ii"), at(p, 29, "ii"), at(p, 37, "ii")] + ] + hole = raw_unit(at(p, 45, "i")) + if hole < 0: raise ParseError("Negative pad drill diameter") + shapes = list(at(p, 49, "BBB")) + rotation = at(p, 52, "d") + mode = at(p, 62, "B") + obj.properties.update( + { + "PAD_NAME": name, + "HOLE_MIL": str(hole * 10), + "ROTATION": str(rotation), + "PLATED": str(bool(p[60])), + "BOTTOM_SIZE_MIL": f"{sizes[2][0]*10:.15g} x {sizes[2][1]*10:.15g}", + "MID_SIZE_MIL": f"{sizes[1][0]*10:.15g} x {sizes[1][1]*10:.15g}", + "TOP_SIZE_MIL": f"{sizes[0][0] * 10:.15g} x {sizes[0][1] * 10:.15g}", + "SHAPE": str(shapes[0]), + "PAD_MODE": str(mode), + } + ) + target_layers = ( + copper_ids if int(obj.layer_id) == 74 else [int(obj.layer_id)] + ) + obj.layer_ids = [str(lid) for lid in target_layers] + for lid in target_layers: + if mode not in (0,1,2) or (mode == 2 and len(stack) < 596): + report("pad_stack", "Unsupported pad stack mode or truncated full-stack variant", obj.id) + continue + which = 0 if lid == 1 else 2 if lid == 32 else 1 + w, h = sizes[which] + s = shapes[which] + radius = 0 + if mode == 2 and len(stack) >= 596 and 2 <= lid <= 30: + w, h = ( + raw_unit(at(stack, 4 * (lid - 2), "i")), + raw_unit(at(stack, 116 + 4 * (lid - 2), "i")), + ) + s = stack[232 + lid - 2] + stack_index = 0 if lid == 1 else 31 if lid == 32 else lid - 1 + if mode == 2 and lid > 32: + report("pad_plane_stack", "Full-stack pad geometry on internal plane layers is not decoded", obj.id) + continue + if len(stack) >= 596 and lid <= 32: + alt = stack[532 + stack_index] + radius = stack[564 + stack_index] + if alt in (1, 2, 3, 9): + s = alt + elif alt != 0: + report( + "pad_alt_shape", + f"Unsupported alternative pad shape {alt}", + obj.id, + ) + continue + try: + path, box = pad_shape(pos, w, h, s, rotation, radius) + except ParseError as exc: + report("pad_shape", str(exc), obj.id) + continue + shape(obj, path, box, fill="layer", layer_id=str(lid), role="pad") + if hole > 0: + hole_shape = stack[262] if len(stack) >= 596 else 0 + slot = raw_unit(at(stack, 263, "i")) if len(stack) >= 596 else 0 + hole_angle = at(stack, 267, "d") if len(stack) >= 596 else 0 + dx = raw_unit(at(stack, 275, "i")) if len(stack) >= 596 else 0 + dy = -raw_unit(at(stack, 403, "i")) if len(stack) >= 596 else 0 + hpos = transform_point(Point(x=dx, y=dy), pos, rotation) + if hole_shape == 2 and slot > hole: + path, box = pad_shape(hpos, slot, hole, 1, rotation + hole_angle) + elif hole_shape in (0, 1): + path, box = pad_shape( + hpos, hole, hole, 2 if hole_shape == 1 else 1, rotation + ) + else: + report( + "drill_shape", + f"Unsupported pad hole shape {hole_shape}", + obj.id, + ) + continue + drill = shape( + obj, + path, + box, + fill="#08151b", + layer_id="81", + role="drill", + stroke="none", + ) + drill.layer_ids = obj.layer_ids + for i, (p,) in typed("Vias6", 3): + obj, c = common("Vias6", i, p) + obj.kind = "via" + pos = xy(*at(p, 13, "ii")) + obj.position = pos + diameter, hole = raw_unit(at(p, 21, "i")), raw_unit(at(p, 25, "i")) + if diameter <= 0 or hole < 0: + raise ParseError("Invalid via diameter") + obj.properties.update( + { + "DIAMETER_MIL": str(diameter * 10), + "HOLE_MIL": str(hole * 10), + "START_LAYER": str(at(p, 29, "B")), + "END_LAYER": str(at(p, 30, "B")), + } + ) + start_layer, end_layer = int(obj.properties["START_LAYER"]), int(obj.properties["END_LAYER"]) + if start_layer not in copper_ids or end_layer not in copper_ids: + raise ParseError("Via span refers to an inactive/unsupported copper layer") + a, b = sorted((copper_ids.index(start_layer), copper_ids.index(end_layer))) + span = copper_ids[a:b+1] + obj.layer_ids = [str(lid) for lid in span] + obj.properties["LAYER_SPAN"] = ",".join(obj.layer_ids) + diameters = at(p,75,"32i") if len(p) >= 203 else None + for lid in span: + if diameters and lid > 32 and len(set(diameters)) > 1: + report("via_plane_stack", "Nonuniform via diameter on internal plane layers is not decoded", obj.id) + continue + size = raw_unit(diameters[lid-1]) if diameters and lid <= 32 else diameter + if size <= 0: raise ParseError("Invalid per-layer via diameter") + shape(obj, ellipse_path(pos.x,pos.y,size/2),bounds([pos],size/2),fill="layer",layer_id=str(lid),role="via") + if hole: + drill = shape(obj,ellipse_path(pos.x,pos.y,hole/2),bounds([pos],hole/2),fill="#08151b",layer_id="81",role="drill",stroke="none") + drill.layer_ids = obj.layer_ids + strings = wide_strings(read("WideStrings6/Data")) + for i, parts in typed("Texts6", 5, 2): + p = parts[0] + obj, c = common("Texts6", i, p, at(p, 226, "I", 0)) + obj.kind = "text" + pos = xy(*at(p, 13, "ii")) + obj.position = pos + height = raw_unit(at(p, 21, "i")) + if height < 0: raise ParseError("Negative text height") + rotation = at(p, 27, "d") + mirror = bool(p[35]) + font = "Arial" + bold = italic = False + is_comment = len(p) >= 123 and bool(p[40]) + is_designator = len(p) >= 123 and bool(p[41]) + if len(p) >= 123: + font = ( + bytes(p[46:110]).decode("utf-16le", errors="replace").split("\0")[0] + or "Arial" + ) + bold = bool(p[44]) + italic = bool(p[45]) + legacy = Reader(parts[1], "PCB text").pascal() if parts[1] else "" + value = ( + strings.get(at(p, 115, "I", 4294967295), legacy) + if len(p) >= 123 + else legacy + ) + if c and is_designator: + c.designator = value or c.designator + if c.parameters.get("NAMEON", "TRUE") == "FALSE": + obj.visible = False + if c and is_comment: + c.value = value + c.comment = value + if c.parameters.get("COMMENTON", "FALSE") == "FALSE": + obj.visible = False + obj.properties.update( + { + "TEXT_HEIGHT": str(height), + "TEXT_MIRRORED": str(mirror), + "TEXT": value, + "ROTATION": str(rotation), + "FONT_NAME": font, + "DESIGNATOR": str(is_designator), + "COMMENT": str(is_comment), + } + ) + if not obj.visible or not value: + continue + if len(p) > 123 and (p[43] == 2 or p[110] or p[123]): + report( + "text_style", + "PCB barcode/inverted text is approximated by readable text", + obj.id, + ) + # Native font substitution is explicitly diagnosed below. Geometry remains vector text. + size = max(height * 1.35, 0.01) + w = max(size * 0.5, len(value) * size * 0.75) + corners = [ + transform_point(Point(x=x, y=y), pos, rotation, mirror) + for x, y in [(0, 0), (w, 0), (w, -size), (0, -size)] + ] + shape( + obj, + "", + bounds(corners), + position=pos, + text=value, + rotation=-rotation, + mirrored=mirror, + font_size=size, + font_family=font, + bold=bold, + italic=italic, + role="text", + ) + for c in components: + if c.id in component_boxes: + c.bounds = union(component_boxes[c.id]) + if not c.primitive_ids: + report( + "empty_footprint", + f"No supported 2D geometry for {c.designator or c.id}", + ) + if len(primitives) > 250000: + raise ParseError("PCB drawable limit exceeded") + # Group layers once; stable within-layer priority keeps pours behind tracks/pads and drills above all. + priority = { + "board": 0, + "pour": 1, + "region": 2, + "fill": 3, + "track": 4, + "polygon_outline": 4, + "pad": 5, + "via": 6, + "text": 7, + "board_cutout": 8, + "drill": 9, + } + primitives.sort( + key=lambda p: ( + 1001 if p.role == "board_cutout" else layers[p.layer_id].order, + priority.get(p.role, 4), + ) + ) + supported = { + "FileHeader", + "FileHeaderSix", + "UniqueIDPrimitiveInformation", + "Board6", + "Components6", + "Nets6", + "Tracks6", + "Arcs6", + "Pads6", + "Vias6", + "Fills6", + "Texts6", + "WideStrings6", + "Regions6", + "ShapeBasedRegions6", + "Polygons6", + } + omitted = [ + s["name"] + for s in streams + if s["size"] + and s["name"].endswith("/Data") + and s["name"].split("/")[0] not in supported + ] + if omitted: + for stream in omitted: + header = read(stream.removesuffix('/Data')+'/Header') + count = at(header,0,'I') if len(header)==4 else 1 + report("uninterpreted_stream", f"Not interpreted for this 2D viewer: {stream}", stream) + grouped[("uninterpreted_stream",f"Not interpreted for this 2D viewer: {stream}")].count = count + report( + "embedded_resources", + "Embedded STEP/3D models and fonts are not loaded or executed. This is a 2D PCB viewer.", + "", + "info", + ) + report( + "pcb_text_metrics", + "PCB stroke/TrueType text uses browser font substitution; glyph metrics and special text styles can differ.", + ) + report( + "pcb_colors", + "Layer colors use a viewer palette; stored layer identity, geometry, net and footprint ownership are retained.", + "", + "info", + ) + report( + "pcb_scope", + "Saved copper fills are shown without repouring or executing design rules. Negative internal planes, mask-rule expansions and 3D bodies are not reconstructed.", + ) + return Document( + schema_version="1.1", + document_type="pcb", + id=document_id, + filename=filename, + metadata={ + "format": "ole", + "header": envelope, + "minor_version": board.get("VERSION"), + "sha256": hashlib.sha256(data).hexdigest(), + "input_bytes": len(data), + "streams": streams, + "record_counts": counts, + "record_count": sum(counts.values()), + "object_count": len(objects), + "record_variants": {k: dict(v) for k, v in variants.items()}, + "component_count": len(components), + "primitive_count": len(primitives), + "units": "10mil", + "coordinates": "PCB integer /100000; X right, Y down", + "compatibility": "partial", + "header_properties": { + "ENVELOPE": envelope, + "BOARD_VERSION": board.get("VERSION", ""), + }, + "net_count": len(nets), + "copper_stack": [str(lid) for lid in copper_ids], + "palette_source": "viewer", + "board_size_mm": { + "width": board_bounds.width * 0.254, + "height": board_bounds.height * 0.254, + }, + "uninterpreted_streams": omitted, + }, + sheet=Sheet( + name=filename, + background="#08151b", + bounds=board_bounds, + properties=board, + ), + bounds=board_bounds, + components=components, + symbols=[], + objects=objects, + primitives=primitives, + diagnostics=diagnostics, + layers=sorted(layers.values(), key=lambda l: l.order, reverse=True), + nets=nets, + ) diff --git a/backend/app/pcb_api.py b/backend/app/pcb_api.py new file mode 100644 index 0000000..9b1f39e --- /dev/null +++ b/backend/app/pcb_api.py @@ -0,0 +1,130 @@ +"""Two PCB creation endpoints using the existing durable queue and capabilities.""" +import hashlib +import json +import re +import secrets +import time +from pathlib import Path +from typing import Annotated, Literal +from fastapi import File, Form, Header, UploadFile +from fastapi.responses import FileResponse, JSONResponse, Response +from sqlalchemy import select, func, text +from starlette.concurrency import run_in_threadpool +from .config import settings +from .db import Job, sessions +from .storage import storage +from .uploads import read_pcb +from .models.pcb import PCBData, PCBStatus, PCBUploadResponse, PCBComparisonStatus, PCBComparisonUploadResponse + +FRONTEND = Path(__file__).resolve().parents[2] / 'frontend' + + +def register_pcb_routes(app, *, authorize, require_api, error): + def state(job, comparison=False): + base = settings().public_url + api = f'/api/v1/pcbs/comparisons/{job.id}' if comparison else f'/api/v1/pcbs/{job.id}' + view = f'/compare/pcb/{job.id}' if comparison else f'/view/pcb/{job.id}' + result = dict(success=job.status != 'failed', id=job.id, status=job.status, + document_type='pcb_comparison' if comparison else 'pcb', + viewer_url=base+view, status_url=base+api+('' if comparison else '/status'), + access=job.access, expires_at=job.expires_at) + if comparison: result.update(comparison_id=job.id, url=base+view) + if job.summary: result.update(json.loads(job.summary)) + if job.error_code: result['error'] = dict(code=job.error_code, message=job.error_message) + return result + + async def create(files, access, expires_in, comparison): + ttl = expires_in or settings().link_ttl + if ttl > settings().link_ttl: error('invalid_expiry','expires_in exceeds configured link lifetime',422) + sources = {} + try: + for field, file in files.items(): sources[field] = await read_pcb(file) + finally: + for file in files.values(): await file.close() + identifier = secrets.token_hex(24); token = secrets.token_urlsafe(32); now = time.time() + input_name = 'pcb-comparison.json' if comparison else 'input.PcbDoc' + job = Job(id=identifier, filename='PCB comparison' if comparison else sources['file'][0], + status='queued', access=access, token_hash=hashlib.sha256(token.encode()).hexdigest(), + created_at=now, updated_at=now, expires_at=now+ttl, attempt=0, input_key=f'{identifier}/{input_name}') + if comparison: + bodies = {f'{identifier}/{side}.PcbDoc':sources[side+'_file'][1] for side in ('old','new')} + bodies[job.input_key] = json.dumps(dict(schema_version='1.0',document_type='pcb_comparison', + old_filename=sources['old_file'][0],new_filename=sources['new_file'][0])).encode() + else: bodies = {job.input_key:sources['file'][1]} + def persist(): + written = [] + try: + with sessions().begin() as db: + if db.bind.dialect.name == 'postgresql': db.execute(text('SELECT pg_advisory_xact_lock(84721309)')) + pending = db.scalar(select(func.count()).select_from(Job).where(Job.status.in_(['queued','processing']),Job.expires_at>now)) + if pending >= settings().max_pending: error('queue_full','Rendering queue is full; try again later',503) + for key, body in bodies.items(): + # Include the attempted key so a write-then-raise backend is also cleaned up. + written.append(key) + try: storage().put(key,body) + except Exception: error('storage_unavailable','Upload storage unavailable',503) + db.add(job); db.flush() + except Exception: + for key in written: + try: storage().delete(key) + except Exception: pass + raise + await run_in_threadpool(persist) + result = state(job,comparison) + if access == 'private': + result['viewer_token'] = token; result['viewer_url'] += '#token='+token + if comparison: result['url'] = result['viewer_url'] + return JSONResponse(result,status_code=202,headers={'Location':result['status_url'],'Retry-After':'1'}) + + @app.post('/api/v1/pcbs/render', status_code=202, response_model=PCBUploadResponse, response_model_exclude_none=True, tags=['PCB']) + async def render(file:Annotated[UploadFile,File()], access:Annotated[Literal['private','unlisted'],Form()]='private', + expires_in:Annotated[int|None,Form(gt=0)]=None, x_api_key:Annotated[str|None,Header()]=None): + require_api(x_api_key) + return await create({'file':file},access,expires_in,False) + + @app.post('/api/v1/pcbs/comparisons',status_code=202,response_model=PCBComparisonUploadResponse,response_model_exclude_none=True,tags=['PCB']) + async def compare(old_file:Annotated[UploadFile,File()],new_file:Annotated[UploadFile,File()], + access:Annotated[Literal['private','unlisted'],Form()]='private', + expires_in:Annotated[int|None,Form(gt=0)]=None,x_api_key:Annotated[str|None,Header()]=None): + require_api(x_api_key) + return await create({'old_file':old_file,'new_file':new_file},access,expires_in,True) + + def get_job(identifier,key,token,comparison=False): + return authorize(identifier,key,token,resource_kind='pcb_comparison' if comparison else 'pcb_render') + + def artifact(job,name): + if job.status == 'failed': error(job.error_code,job.error_message,422) + if job.status != 'completed': error('not_ready','PCB processing is still in progress',409) + try: body = storage().get(job.output_prefix+'/'+name) + except Exception: error('storage_unavailable','PCB artifact unavailable',503) + return Response(body,media_type='application/json' if name.endswith('.json') else 'image/svg+xml') + + @app.get('/api/v1/pcbs/comparisons/{identifier}',response_model=PCBComparisonStatus,response_model_exclude_none=True,tags=['PCB']) + def comparison_status(identifier:str,x_api_key:Annotated[str|None,Header()]=None,x_viewer_token:Annotated[str|None,Header()]=None): + job=get_job(identifier,x_api_key,x_viewer_token,True); result=state(job,True) + if job.status == 'completed': + try: result.update(json.loads(storage().get(job.output_prefix+'/data.json'))) + except Exception: error('storage_unavailable','PCB comparison unavailable',503) + return result + + @app.get('/api/v1/pcbs/comparisons/{identifier}/svg/{side}',response_class=Response,tags=['PCB']) + def comparison_svg(identifier:str,side:Literal['old','new'],x_api_key:Annotated[str|None,Header()]=None,x_viewer_token:Annotated[str|None,Header()]=None): + return artifact(get_job(identifier,x_api_key,x_viewer_token,True),side+'.svg') + + @app.get('/api/v1/pcbs/{identifier}/status',response_model=PCBStatus,response_model_exclude_none=True,tags=['PCB']) + def status(identifier:str,x_api_key:Annotated[str|None,Header()]=None,x_viewer_token:Annotated[str|None,Header()]=None): + return state(get_job(identifier,x_api_key,x_viewer_token)) + + @app.get('/api/v1/pcbs/{identifier}/data',response_model=PCBData,tags=['PCB']) + def data(identifier:str,x_api_key:Annotated[str|None,Header()]=None,x_viewer_token:Annotated[str|None,Header()]=None): + return artifact(get_job(identifier,x_api_key,x_viewer_token),'data.json') + + @app.get('/api/v1/pcbs/{identifier}/svg',response_class=Response,tags=['PCB']) + def svg(identifier:str,x_api_key:Annotated[str|None,Header()]=None,x_viewer_token:Annotated[str|None,Header()]=None): + return artifact(get_job(identifier,x_api_key,x_viewer_token),'drawing.svg') + + @app.get('/view/pcb/{identifier}',include_in_schema=False) + @app.get('/compare/pcb/{identifier}',include_in_schema=False) + def page(identifier:str): + if not re.fullmatch('[a-f0-9]{48}',identifier): error('not_found','PCB not found',404) + return FileResponse(FRONTEND/'pcb.html') diff --git a/backend/app/pcb_diff.py b/backend/app/pcb_diff.py new file mode 100644 index 0000000..0ad7fcb --- /dev/null +++ b/backend/app/pcb_diff.py @@ -0,0 +1,134 @@ +"""PCB object comparison; reuses only the format-independent entity matcher.""" +import math +import re +from collections import defaultdict +from .diff import Entity, digest, match_entities + +TOKEN = re.compile(r'[MLAZ]|[-+]?(?:\d*\.\d+|\d+\.?\d*)(?:[eE][-+]?\d+)?') +IGNORE = {'NET_INDEX','COMPONENT_INDEX','POLYGON_INDEX','INDEX','UNIQUEID','SOURCEUNIQUEID', + 'LOCKED','FLAGS_1','FLAGS_2','OUTLINE_PATH','X','Y','ROTATION','LAYER'} + + +def number(n): + return round(float(n), 7) or 0.0 + + +def point(x, y, origin, rotation=0, mirrored=False): + x,y = x-origin[0],y-origin[1] + a=math.radians(rotation) + x,y=x*math.cos(a)-y*math.sin(a),x*math.sin(a)+y*math.cos(a) + return (number(-x if mirrored else x),number(y)) + + +def path_signature(path, origin, rotation=0, mirrored=False): + """Canonical local geometry for the M/L/A/Z vocabulary emitted by the PCB parser.""" + tokens=TOKEN.findall(path); out=[]; i=0 + while i bytes: b = doc.sheet.bounds + pcb = doc.document_type == 'pcb' + layer_visibility = {layer.id: layer.visible for layer in doc.layers} root = Element('svg', {'xmlns': 'http://www.w3.org/2000/svg', 'viewBox': f'{b.x} {b.y} {b.width} {b.height}', - 'preserveAspectRatio': 'xMidYMid meet', 'role': 'img', 'aria-label': 'Schematic'}) + 'preserveAspectRatio': 'xMidYMid meet', 'role': 'img', 'aria-label': 'PCB' if pcb else 'Schematic'}) node(root, 'rect', x=b.x, y=b.y, width=b.width, height=b.height, fill=doc.sheet.background) defs = node(root, 'defs') for i, p in enumerate(doc.primitives): @@ -20,6 +22,12 @@ def svg_document(doc: Document, annotations=None, id_prefix="") -> bytes: if annotations is not None: annotation = annotations.get(p.component_id or p.object_id) target = node(root, 'g', data_object_id=p.component_id or p.object_id, data_change_id=annotation['id'] if annotation else '') + if pcb: + span = getattr(p, "layer_ids", []) + visible = layer_visibility.get(p.layer_id, True) and (p.role != "drill" or not span or any(layer_visibility.get(lid, False) for lid in span)) + target = node(target, 'g', data_layer_id=p.layer_id or '', + data_layer_span=' '.join(getattr(p, 'layer_ids', [])), data_role=p.role, + display='inline' if visible else 'none') if p.clip: clip = node(defs, 'clipPath', id=f'{id_prefix}clip-{i}', clipPathUnits='userSpaceOnUse') c = p.clip diff --git a/backend/app/storage.py b/backend/app/storage.py index 1851fe4..7d88d4c 100644 --- a/backend/app/storage.py +++ b/backend/app/storage.py @@ -6,7 +6,7 @@ from functools import lru_cache from pathlib import Path from .config import settings -KEY = re.compile(r'^[a-f0-9]{48}/(?:input\.SchDoc|(?:old|new)\.SchDoc|comparison\.json|attempt-[12]/(?:data\.json|drawing\.svg|old\.svg|new\.svg))$') +KEY = re.compile(r'^[a-f0-9]{48}/(?:input\.(?:SchDoc|PcbDoc)|(?:old|new)\.(?:SchDoc|PcbDoc)|comparison\.json|pcb-comparison\.json|attempt-[12]/(?:data\.json|drawing\.svg|old\.svg|new\.svg))$') class Storage: def __init__(self): diff --git a/backend/app/uploads.py b/backend/app/uploads.py index 0c37ed9..a37e739 100644 --- a/backend/app/uploads.py +++ b/backend/app/uploads.py @@ -3,18 +3,36 @@ from fastapi import HTTPException from .config import settings from .parser.record_decoder import ParseError,detect_format -async def read_schematic(file): +async def read_native(file, document_type): try: name=(file.filename or '').replace('\\','/').rsplit('/',1)[-1] name=''.join(c for c in name if c.isprintable())[:255] - if not name.lower().endswith('.schdoc'): - raise HTTPException(415,{'code':'unsupported_file_type','message':'Upload a .SchDoc file'}) + suffix = '.PcbDoc' if document_type == 'pcb' else '.SchDoc' + if not name.lower().endswith(suffix.lower()): + raise HTTPException(415,{'code':'unsupported_file_type','message':f'Upload a {suffix} file'}) body=bytearray() while chunk:=await file.read(1024*1024): body.extend(chunk) if len(body)>settings().max_upload_bytes: raise HTTPException(413,{'code':'upload_too_large','message':'Schematic exceeds configured upload size'}) - try:detect_format(bytes(body)) - except ParseError as exc:raise HTTPException(422,{'code':'invalid_schematic','message':str(exc)}) from exc + from starlette.concurrency import run_in_threadpool + from .parser.pcb_container import identify_pcb, contains_pcb + try: + if document_type == 'pcb': + await run_in_threadpool(identify_pcb, bytes(body)) + else: + detect_format(bytes(body)) + if await run_in_threadpool(contains_pcb, bytes(body)): + raise ParseError('PCB files are not accepted by schematic endpoints') + except ParseError as exc: + raise HTTPException(422,{'code':'invalid_pcb' if document_type == 'pcb' else 'invalid_schematic','message':str(exc)}) from exc return name,bytes(body) finally:await file.close() + + +async def read_schematic(file): + return await read_native(file, "schematic") + + +async def read_pcb(file): + return await read_native(file, "pcb") diff --git a/backend/app/worker.py b/backend/app/worker.py index eaae2d3..1f20cb3 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -13,6 +13,7 @@ from .config import settings from .cleanup import cleanup from .db import Job, claim, initialize, sessions from .storage import storage +from .job_types import job_kind, comparison_sources log = logging.getLogger(__name__) stopping = False @@ -23,19 +24,24 @@ def process(job): try: with tempfile.TemporaryDirectory(prefix='schdoc-') as directory: root = Path(directory) - source = root / 'input.SchDoc' - comparison = job['input_key'].endswith('/comparison.json') + kind = job_kind(job['input_key']) + if kind is None: raise ValueError('Unknown job input kind') + pcb = kind.startswith('pcb_') + source = root / ('input.PcbDoc' if pcb else 'input.SchDoc') + comparison = kind.endswith('_comparison') if comparison: manifest = json.loads(storage().get(job['input_key'])) source = root / 'comparison.json' source.write_text(json.dumps(manifest), encoding='utf-8') - for side in ('old','new'): - (root / f'{side}.SchDoc').write_bytes(storage().get(f"{job['id']}/{side}.SchDoc")) + for key in comparison_sources(job['input_key']): + (root / key.rsplit('/', 1)[-1]).write_bytes(storage().get(key)) else: source.write_bytes(storage().get(job['input_key'])) env = {k: v for k, v in os.environ.items() if k in ('PATH', 'LANG', 'LC_ALL', 'PYTHONPATH')} # Drain no untrusted output into logs. Kill/wait is handled by subprocess.run. - subprocess.run([sys.executable, '-m', 'app.compare_task' if comparison else 'app.parse_task', str(source), str(root), job['filename'], job['id']], + task = 'app.pcb_task' if pcb else 'app.compare_task' if comparison else 'app.parse_task' + args = [kind] if pcb else [] + subprocess.run([sys.executable, '-m', task, str(source), str(root), job['filename'], job['id'], *args], timeout=config.parse_timeout, check=True, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) result = json.loads((root / 'result.json').read_text()) if result['ok']: diff --git a/backend/tests/pcb_fixture.py b/backend/tests/pcb_fixture.py new file mode 100644 index 0000000..e274c4e --- /dev/null +++ b/backend/tests/pcb_fixture.py @@ -0,0 +1,99 @@ +"""Tiny CFB writer solely for bounded native-format unit fixtures (never application data).""" +import math +import struct + +FREE=0xffffffff; END=0xfffffffe; FAT=0xfffffffd + + +def cfb(streams): + entries=[('Root Entry',5,b'',[])] + storages={} + for name,data in streams.items(): + parts=name.split('/'); parent=0 + for part in parts[:-1]: + key=(parent,part) + if key not in storages: + storages[key]=len(entries);entries.append((part,1,b'',[]));entries[parent][3].append(storages[key]) + parent=storages[key] + entries[parent][3].append(len(entries));entries.append((parts[-1],2,data,[])) + # This fixture deliberately uses normal >=4096-byte streams, with padding INSIDE native records. + assert all(len(e[2])>=4096 for e in entries if e[1]==2) + sectors=[];links=[];starts={} + def allocate(data): + start=len(sectors); count=math.ceil(len(data)/512) + for i in range(count):sectors.append(data[i*512:(i+1)*512].ljust(512,b'\0'));links.append(start+i+1 if i+10 and len(doc.primitives)>0 + assert len({p.id for p in doc.primitives})==len(doc.primitives) + assert compare_pcbs(doc,doc)['total_changed']==0 + assert any(p.role=='pad' for p in doc.primitives) + assert any(p.role=='drill' for p in doc.primitives) + assert all(set(o.layer_ids)<={layer.id for layer in doc.layers} for o in doc.objects) + + +def test_pad_stack_slot_via_layers_and_native_ids(): + from pcb_fixture import rich_pcb + doc=parse(rich_pcb());pad=next(o for o in doc.objects if o.kind=='pad');via=next(o for o in doc.objects if o.kind=='via') + assert pad.native_uid=='PAD_UID' and pad.component_id==doc.components[0].id + assert pad.layer_ids==['1','32'] and via.layer_ids==['1','32'] + pads=[p for p in doc.primitives if p.object_id==pad.id and p.role=='pad'] + assert {p.layer_id for p in pads}=={'1','32'} + assert next(p for p in pads if p.layer_id=='1').bounds.width==1 + assert next(p for p in pads if p.layer_id=='32').bounds.width==2 + slot=next(p for p in doc.primitives if p.object_id==pad.id and p.role=='drill') + assert slot.bounds.width==1 and slot.bounds.height==.5 and slot.layer_ids==['1','32'] + bottom_via=next(p for p in doc.primitives if p.object_id==via.id and p.layer_id=='32') + assert bottom_via.bounds.width==2 + + +def test_footprint_move_rotation_and_child_edits_are_one_change(): + from pcb_fixture import rich_pcb + from app.parser.pcb_geometry import pad_shape + old=parse(rich_pcb());new=old.model_copy(deep=True) + component=new.components[0];component.position.x+=4 + for o in new.objects: + if o.component_id==component.id:o.position.x+=4;o.bounds.x+=4 + # Paths translated independently from their existing SVG string. + for p in new.primitives: + if p.component_id==component.id: + if p.role=='pad': + w,h=(1,2) if p.layer_id=='1' else (2,3) + p.path,p.bounds=pad_shape(Point(x=5,y=-1),w,h,1 if p.layer_id=='1' else 3,0) + elif p.role=='drill':p.path,p.bounds=pad_shape(Point(x=5,y=-1),1,.5,1,0) + # Source parameters retain physical placement too. + component.parameters['X']='50mil' + result=compare_pcbs(old,new) + assert result['summary']==dict(added=0,removed=0,modified=0,moved=1),result['changes'] + component.value='new value' + result=compare_pcbs(old,new) + assert result['summary']['modified']==1 and result['changes'][0]['moved'] + # Reparse native revision with a rotated footprint/pad: only one footprint modification. + rotated=parse(rich_pcb());rotated.components[0].rotation=90 + assert compare_pcbs(old,rotated)['summary']['modified']==1 + + +def test_record_reordering_net_renumber_and_duplicate_ambiguity(): + import io + import olefile + from pcb_fixture import cfb,text,record + def native(reverse): + with olefile.OleFileIO(io.BytesIO(pcb())) as ole:streams={'/'.join(e):ole.openstream(e).read() for e in ole.listdir()} + streams['Nets6/Data']=text({'NAME':'VCC'})+text({'NAME':'GND'}) if reverse else text({'NAME':'GND'})+text({'NAME':'VCC'}) + records=[] + for x,net in [(100000,0),(900000,1)]: + p=bytearray(45);p[0]=1 + struct.pack_into(' selected.has(layer)); +} +export function presetLayers(layers, preset) { + return new Set(layers.filter(layer => preset === 'all' || layer.id === '0' || layer.id === '81' || + (preset === 'top' ? ['1','33'].includes(layer.id) : ['32','34'].includes(layer.id))).map(layer => layer.id)); +} diff --git a/frontend/pcb-layers.test.js b/frontend/pcb-layers.test.js new file mode 100644 index 0000000..0a39f68 --- /dev/null +++ b/frontend/pcb-layers.test.js @@ -0,0 +1,15 @@ +import assert from 'node:assert/strict'; +import {test} from 'node:test'; +import {layerVisible,presetLayers} from './pcb-layers.js'; +test('blind via drills follow actual selected copper span',()=>{ + assert.equal(layerVisible('81',['2','3'],new Set(['1','81'])),false); + assert.equal(layerVisible('81',['2','3'],new Set(['2','81'])),true); + assert.equal(layerVisible('81',['1'],new Set(['1'])),false); + assert.equal(layerVisible('0',[],new Set(['0'])),true); +}); +test('layer presets preserve native IDs and outline',()=>{ + const layers=['0','1','2','32','33','34','81'].map(id=>({id})); + assert.deepEqual([...presetLayers(layers,'top')],['0','1','33','81']); + assert.deepEqual([...presetLayers(layers,'bottom')],['0','32','34','81']); + assert.equal(presetLayers(layers,'all').size,7); +}); diff --git a/frontend/pcb.css b/frontend/pcb.css new file mode 100644 index 0000000..3b9e4f0 --- /dev/null +++ b/frontend/pcb.css @@ -0,0 +1,20 @@ +[hidden] { display: none !important; } +#pcb-workspace { flex: 1; min-height: 0; display: grid; grid-template-columns: minmax(0,1fr); } +body.pcb-comparison #pcb-workspace { grid-template-columns: minmax(0,1fr) minmax(0,1fr); } +body.menu-open #pcb-workspace { grid-template-columns: minmax(0,1fr) 270px; } +body.pcb-comparison.menu-open #pcb-workspace { grid-template-columns: minmax(0,1fr) minmax(0,1fr) 270px; } +#pcb-menu { padding: 10px; overflow: auto; font-size: 12px; } +#pcb-menu h2 { padding: 5px 0; background: transparent; } +#layers { display: grid; gap: 5px; margin: 12px 0; } +#layer-presets { display: flex; gap: 5px; } +#compatibility { color: #60716b; font-size: 11px; overflow-wrap: anywhere; } +#diff-controls { min-height: 100px; } +#changes { max-height: 35vh; padding: 0; margin-top: 10px; } +#summary { font-size: 12px; } +.surface { background: #08151b; } +@media (max-width: 700px) { + body.pcb-comparison #pcb-workspace { grid-template-columns: 1fr; grid-template-rows: minmax(0,1fr) minmax(0,1fr); } + body.menu-open #pcb-workspace { grid-template-columns: minmax(0,1fr) 210px; } + body.pcb-comparison.menu-open #pcb-workspace { grid-template-columns: minmax(0,1fr) 210px; grid-template-rows: minmax(0,1fr) minmax(0,1fr); } + #pcb-menu { grid-column: 2; grid-row: 1 / -1; } +} diff --git a/frontend/pcb.html b/frontend/pcb.html new file mode 100644 index 0000000..13ac84e --- /dev/null +++ b/frontend/pcb.html @@ -0,0 +1,5 @@ + +PCB Viewer + + +
Loading PCB… diff --git a/frontend/pcb.js b/frontend/pcb.js new file mode 100644 index 0000000..0677f71 --- /dev/null +++ b/frontend/pcb.js @@ -0,0 +1,85 @@ +import { bindCamera } from './interactions.js'; +import { layerVisible, presetLayers } from './pcb-layers.js'; +const comparison=location.pathname.startsWith('/compare/pcb/'); +const id=location.pathname.split('/').pop(), state=document.querySelector('#state'); +const key=`pcb:${comparison?'comparison':'render'}:${id}`; +const token=new URLSearchParams(location.hash.slice(1)).get('token')||sessionStorage.getItem(key); +if(token)sessionStorage.setItem(key,token); +if(location.hash)history.replaceState(null,'',location.pathname); +const headers=token?{'X-Viewer-Token':token}:{}, controller=new AbortController(), views=[]; +const base=comparison?`/api/v1/pcbs/comparisons/${id}`:`/api/v1/pcbs/${id}`; +const menu=document.querySelector('#menu-toggle'); +function setMenu(open){ + document.body.classList.toggle('menu-open',open); + for(const panel of document.querySelectorAll('[data-pcb-menu]'))panel.hidden=!open; + menu.setAttribute('aria-expanded',String(open));menu.setAttribute('aria-label',open?'Hide menu':'Show menu'); +} +setMenu(false); +menu.addEventListener('click',()=>setMenu(menu.getAttribute('aria-expanded')!=='true')); +addEventListener('keydown',e=>{if(e.key==='Escape'){setMenu(false);menu.focus();}}); +addEventListener('pagehide',()=>{controller.abort();for(const view of views)view.destroy();},{once:true}); +document.body.classList.toggle('pcb-comparison',comparison); +document.querySelector('#new-pane').hidden=!comparison; +for(const selector of ['#sync-control','#highlight-control','#diff-controls'])document.querySelector(selector).hidden=!comparison; +async function request(path='',text=false){ + const r=await fetch(base+path,{headers,signal:controller.signal,cache:'no-store'}); + if(!r.ok){const b=await r.json();throw new Error(b.error?.message||'PCB unavailable');} + return text?r.text():r.json(); +} +function union(boxes){ + const x=Math.min(...boxes.map(b=>b.x)),y=Math.min(...boxes.map(b=>b.y)); + return {x,y,width:Math.max(...boxes.map(b=>b.x+b.width))-x,height:Math.max(...boxes.map(b=>b.y+b.height))-y}; +} +async function start(){ + let result; + for(;;){result=await request(comparison?'':'/status');if(result.status==='completed')break;if(result.status==='failed')throw new Error(result.error?.message||'PCB processing failed');await new Promise(r=>setTimeout(r,1000));} + const texts=await Promise.all((comparison?['/svg/old','/svg/new']:['/svg']).map(path=>request(path,true))); + const svgs=texts.map(text=>{ + const xml=new DOMParser().parseFromString(text,'image/svg+xml'); + if(xml.querySelector('parsererror')||xml.documentElement.localName!=='svg')throw new Error('Invalid PCB artifact'); + return document.importNode(xml.documentElement,true); + }); + const boxes=comparison?[result.old.bounds,result.new.bounds]:svgs.map(svg=>{const b=svg.viewBox.baseVal;return {x:b.x,y:b.y,width:b.width,height:b.height};}); + const bounds=union(boxes); + for(const [i,svg]of svgs.entries()){ + const surface=document.querySelector(i?'#new-surface':'#old-surface');surface.append(svg); + views.push(bindCamera(surface,svg,bounds,source=>{if(comparison&&document.querySelector('#sync').checked)for(const view of views)if(view!==source)view.apply(source.pose());})); + } + await document.fonts.ready; + state.classList.remove('loading');state.textContent=''; + document.querySelector('#fit').addEventListener('click',()=>{for(const view of views)view.fit();}); + document.querySelector('#sync').addEventListener('change',()=>{if(views[1]&&document.querySelector('#sync').checked)views[1].apply(views[0].pose());}); + const layers=comparison?[...new Map([...result.old.layers,...result.new.layers].map(l=>[l.id,l])).values()]:result.layers; + let selected=new Set(layers.filter(l=>l.visible).map(l=>l.id)); + const layerNodes=svgs.flatMap(svg=>[...svg.querySelectorAll('[data-layer-id]')]); + function paintLayers(){for(const node of layerNodes)node.setAttribute('display',layerVisible(node.dataset.layerId,(node.dataset.layerSpan||'').split(' ').filter(Boolean),selected)?'inline':'none');} + const checks=new Map(); + for(const layer of layers){ + const label=document.createElement('label'),input=document.createElement('input');input.type='checkbox';input.checked=selected.has(layer.id); + checks.set(layer.id,input);label.append(input,document.createTextNode(layer.name));document.querySelector('#layers').append(label); + input.addEventListener('change',()=>{if(input.checked)selected.add(layer.id);else selected.delete(layer.id);paintLayers();}); + } + for(const button of document.querySelectorAll('[data-preset]'))button.addEventListener('click',()=>{selected=presetLayers(layers,button.dataset.preset);for(const [id,input]of checks)input.checked=selected.has(id);paintLayers();}); + paintLayers(); + const diagnostics=comparison?[...result.old.diagnostics,...result.new.diagnostics]:result.diagnostics||[]; + document.querySelector('#compatibility').textContent=[...new Set(diagnostics.map(d=>d.message))].join('\n'); + if(comparison){ + const changes=result.changes,filter=document.querySelector('#filter'),rows=new Map();let active=null; + document.querySelector('#summary').textContent=Object.entries(result.summary).map(([k,v])=>`${v} ${k}`).join(' · '); + function focus(change){active=change.id;const boxes=[change.old_bounds,change.new_bounds].filter(Boolean);if(boxes.length)for(const view of views)view.focus(union(boxes));document.querySelector('#details').textContent=JSON.stringify({kind:change.kind,match:change.match,movement:change.movement,fields:change.fields},null,2);for(const [id,row]of rows)row.setAttribute('aria-current',String(id===active));} + for(const change of changes){const button=document.createElement('button');button.type='button';button.className=change.status;button.textContent=`${change.status} · ${change.label}`;button.addEventListener('click',()=>focus(change));document.querySelector('#changes').append(button);rows.set(change.id,button);} + if(!changes.length)document.querySelector('#changes').textContent='No structural changes detected.'; + const byId=new Map(changes.map(c=>[c.id,c])),nodes=svgs.flatMap(svg=>[...svg.querySelectorAll('[data-change-id]')]); + function apply(){ + for(const node of nodes){const c=byId.get(node.dataset.changeId);node.toggleAttribute('data-diff-hidden',c?(filter.value!=='all'&&filter.value!==c.status):(filter.value!=='all'||document.querySelector('#only-changes').checked));} + for(const change of changes)rows.get(change.id).hidden=filter.value!=='all'&&filter.value!==change.status; + document.body.classList.toggle('no-highlights',!document.querySelector('#highlights').checked); + } + for(const input of [filter,document.querySelector('#only-changes'),document.querySelector('#highlights')])input.addEventListener('change',apply); + for(const [selector,step]of [['#previous',-1],['#next',1]])document.querySelector(selector).addEventListener('click',()=>{const visible=changes.filter(c=>filter.value==='all'||c.status===filter.value);if(!visible.length)return;const i=visible.findIndex(c=>c.id===active);focus(visible[(i<0?(step>0?0:visible.length-1):(i+step+visible.length)%visible.length)]);}); + for(const svg of svgs)svg.addEventListener('click',e=>{const c=byId.get(e.target.closest('[data-change-id]')?.dataset.changeId);if(c)focus(c);}); + apply(); + } + function expire(){const remaining=result.expires_at*1000-Date.now();if(remaining<=0){for(const view of views)view.destroy();document.querySelector('#pcb-workspace').replaceChildren();state.textContent='This PCB link has expired';}else setTimeout(expire,Math.min(2147483647,remaining));}expire(); +} +start().catch(e=>{state.classList.remove('loading');state.textContent=e.name==='AbortError'?'':e.message;}); diff --git a/scripts/pcb-smoke.py b/scripts/pcb-smoke.py new file mode 100644 index 0000000..50a6299 --- /dev/null +++ b/scripts/pcb-smoke.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""Real HTTP smoke: private PCB render + comparison, without printing read capabilities.""" +import argparse +import json +import os +from pathlib import Path +import secrets +import time +import urllib.error +import urllib.request + + +def main(): + parser=argparse.ArgumentParser(description=__doc__) + parser.add_argument('old',type=Path);parser.add_argument('new',type=Path,nargs='?') + parser.add_argument('--url',default='http://127.0.0.1:8000') + parser.add_argument('--timeout',type=int,default=180) + args=parser.parse_args();base=args.url.rstrip('/');key=os.environ.get('API_KEY','') + def request(path,headers=None,data=None): + req=urllib.request.Request(base+path,data=data,headers=headers or {}) + with urllib.request.urlopen(req,timeout=30) as response:return response.status,response.headers,response.read() + def upload(path,files): + boundary='pcb-'+secrets.token_hex(16);parts=[] + for field,file in files.items(): + parts.extend([f'--{boundary}\r\nContent-Disposition: form-data; name="{field}"; filename="board.PcbDoc"\r\nContent-Type: application/octet-stream\r\n\r\n'.encode(),file.read_bytes(),b'\r\n']) + parts.extend([f'--{boundary}\r\nContent-Disposition: form-data; name="access"\r\n\r\nprivate\r\n--{boundary}--\r\n'.encode()]) + status,headers,body=request(path,{'X-API-Key':key,'Content-Type':f'multipart/form-data; boundary={boundary}'},b''.join(parts)) + assert status==202;created=json.loads(body);assert '#token=' in created['viewer_url'] + return created + for comparison in (False,True): + path='/api/v1/pcbs/comparisons' if comparison else '/api/v1/pcbs/render' + files={'old_file':args.old,'new_file':args.new or args.old} if comparison else {'file':args.old} + created=upload(path,files);identifier=created['id'];headers={'X-Viewer-Token':created['viewer_token']} + resource=f'/api/v1/pcbs/comparisons/{identifier}' if comparison else f'/api/v1/pcbs/{identifier}' + status_path=resource if comparison else resource+'/status' + try:request(status_path) + except urllib.error.HTTPError as exc:assert exc.code==404 + else:raise AssertionError('Private PCB accessible without a capability') + deadline=time.monotonic()+args.timeout + while True: + _,_,body=request(status_path,headers);result=json.loads(body) + if result['status']=='failed':raise RuntimeError(result['error']) + if result['status']=='completed':break + if time.monotonic()>deadline:raise TimeoutError('PCB worker did not finish') + time.sleep(.5) + for side in (('old','new') if comparison else ('',)): + suffix='/svg/'+side if comparison else '/svg' + status,_,svg=request(resource+suffix,headers);assert status==200 and b'