"""Structural SchDoc comparison. Physical record indexes are never cross-file identity.""" import hashlib import json import re from collections import defaultdict from dataclasses import dataclass from .models.schematic import Document BOOKKEEPING = {'OWNERINDEX','OWNERINDEXADDITIONALLIST','INDEXINSHEET','UNIQUEID', 'GRAPHICALLYLOCKED','LOCKED','READONLYSTATE','_RESOLVED_TEXT'} X_COORD = re.compile(r'^(LOCATION\.X|CORNER\.X|E?X\d+)$') Y_COORD = re.compile(r'^(LOCATION\.Y|CORNER\.Y|E?Y\d+)$') NET_TYPES = {17,18,25} def digest(value): return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(',',':')).encode()).hexdigest() def normalized(properties, origin): result = {} keys = set(properties) | {k[:-5] for k in properties if k.endswith('_FRAC') and (X_COORD.match(k[:-5]) or Y_COORD.match(k[:-5]))} for key in sorted(keys): value = properties.get(key, '0') if key in BOOKKEEPING or (key.endswith('_FRAC') and (X_COORD.match(key[:-5]) or Y_COORD.match(key[:-5]))): continue if X_COORD.match(key) or Y_COORD.match(key): try: offset = origin[0] if X_COORD.match(key) else -origin[1] result[key] = round(float(value) + float(properties.get(key+'_FRAC',0))/100000 - offset, 8) except ValueError: result[key] = value else: result[key] = value return result @dataclass class Entity: id: str kind: str uid: str reference: str position: tuple bounds: dict | None properties: dict signature: str primitive_ids: list record_type: int def entities(document): objects = {o.id:o for o in document.objects} children = defaultdict(list) for obj in document.objects: if obj.component_id: children[obj.component_id].append(obj) image_hashes = defaultdict(list) for primitive in document.primitives: if primitive.kind == 'image': image_hashes[primitive.object_id].append(digest(primitive.image_data)) output = [] for component in document.components: source = objects.get(component.id) origin = (component.position.x, component.position.y) records = [] for obj in children[component.id]: record = normalized(obj.properties, origin) if image_hashes[obj.id]: record['_IMAGE_SHA256'] = image_hashes[obj.id] records.append(record) records.sort(key=digest) details = {key:getattr(component,key) for key in ('designator','value','comment','library_reference', 'source_library','description','rotation','mirrored','current_part','parameters')} details['symbol_records'] = digest(records) output.append(Entity(component.id,'component',source.properties.get('UNIQUEID','').strip().upper() if source else '', f'{component.designator.upper()}:{component.current_part}' if component.designator else '', origin,component.bounds.model_dump() if component.primitive_ids else None, details,digest(details),component.primitive_ids,1)) for obj in document.objects: if obj.component_id or obj.record_type in {31,39,44,45,46,47,48}: continue if not obj.visible and obj.record_type != 41: continue origin = (obj.position.x,obj.position.y) if 'LOCATION.X' not in obj.properties and 'LOCATION.Y' not in obj.properties and obj.connection_points: origin = (obj.connection_points[0].x,obj.connection_points[0].y) raw = obj.properties if obj.record_type in (26,27) and obj.connection_points: points = [(p.x,p.y) for p in obj.connection_points] points = min(points, list(reversed(points))) origin = points[0] raw = {k:v for k,v in raw.items() if not re.fullmatch(r'[XY]\d+(?:_FRAC)?',k)} for i,(x,y) in enumerate(points,1): raw.update({f'X{i}':str(x),f'Y{i}':str(-y)}) properties = normalized(raw, origin) if image_hashes[obj.id]: properties['_IMAGE_SHA256'] = image_hashes[obj.id] output.append(Entity(obj.id,obj.kind,obj.properties.get('UNIQUEID','').strip().upper(),'',origin, obj.bounds.model_dump() if obj.primitive_ids else None,properties,digest(properties),obj.primitive_ids,obj.record_type)) return output def match_entities(old, new): left = {e.id:e for e in old}; right = {e.id:e for e in new} pairs = []; ambiguous = 0 # Stages only pair keys unique on BOTH sides. Coincident/duplicate objects # are not assigned arbitrary nearest neighbors or physical record indexes. stages = [ ('unique_id', lambda e:(e.kind,e.uid) if e.uid else None), ('designator', lambda e:(e.kind,e.reference) if e.reference else None), ('exact', lambda e:(e.kind,e.signature,e.position)), ('structure', lambda e:(e.kind,e.signature)), ('position', lambda e:(e.kind,e.position) if e.bounds is not None else None), ] for method,key in stages: a=defaultdict(list); b=defaultdict(list) for e in left.values(): k=key(e) if k is not None:a[k].append(e) for e in right.values(): k=key(e) if k is not None:b[k].append(e) for k,values in a.items(): candidates=b.get(k,[]) if len(values)==len(candidates)==1: x,y=values[0],candidates[0] pairs.append((x,y,method));del left[x.id];del right[y.id] # Identical duplicates are a multiset, so reordering them is not a change. a=defaultdict(list);b=defaultdict(list) for e in left.values():a[(e.kind,e.signature,e.position)].append(e) for e in right.values():b[(e.kind,e.signature,e.position)].append(e) for key,values in a.items(): for x,y in zip(values,b.get(key,[])): pairs.append((x,y,'identical_multiset'));del left[x.id];del right[y.id] a=defaultdict(int); b=defaultdict(int) for entity in left.values(): a[(entity.kind,entity.signature)]+=1 for entity in right.values(): b[(entity.kind,entity.signature)]+=1 ambiguous=sum(1 for k in a.keys() & b.keys() if a[k]>1 or b[k]>1) return pairs,list(left.values()),list(right.values()),ambiguous def compare_documents(old: Document, new: Document): if old.document_type != 'schematic' or new.document_type != 'schematic': raise ValueError('Only schematic documents can be compared') before,after = entities(old),entities(new) pairs,removed,added,ambiguous = match_entities(before,after) changes=[] def change(status,a,b,method='',fields=None): changes.append({'id':f'change-{len(changes)}','status':status,'kind':(b or a).kind, 'label':(b or a).properties.get('designator') or (b or a).properties.get('TEXT') or (b or a).properties.get('NAME') or (b or a).kind, 'old_id':a.id if a else None,'new_id':b.id if b else None, 'old_bounds':a.bounds if a else None,'new_bounds':b.bounds if b else None, 'old_primitive_ids':a.primitive_ids if a else [],'new_primitive_ids':b.primitive_ids if b else [], 'match':method,'moved':bool(a and b and a.position!=b.position), 'movement':{'dx':b.position[0]-a.position[0],'dy':b.position[1]-a.position[1]} if a and b else None, 'fields':fields or []}) for a,b,method in pairs: fields=[{'property':key,'old':a.properties.get(key),'new':b.properties.get(key)} for key in sorted(a.properties.keys()|b.properties.keys()) if a.properties.get(key)!=b.properties.get(key)] moved=a.position!=b.position if fields or moved:change('modified' if fields else 'moved',a,b,method,fields) for a in removed:change('removed',a,None) for b in added:change('added',None,b) sheet_fields=[{'property':key,'old':old.sheet.properties.get(key),'new':new.sheet.properties.get(key)} for key in sorted(old.sheet.properties.keys()|new.sheet.properties.keys()) if key not in BOOKKEEPING and old.sheet.properties.get(key)!=new.sheet.properties.get(key)] if sheet_fields: changes.append({'id':f'change-{len(changes)}','status':'modified','kind':'sheet','label':'Sheet settings', 'old_id':old.sheet.id,'new_id':new.sheet.id,'old_bounds':old.sheet.bounds.model_dump(), 'new_bounds':new.sheet.bounds.model_dump(),'old_primitive_ids':[],'new_primitive_ids':[], 'match':'sheet','moved':False,'movement':None,'fields':sheet_fields}) def nets(items): result=defaultdict(list) for e in items: if e.record_type in NET_TYPES: name=e.properties.get('TEXT',e.properties.get('NAME','')).strip() if name:result[name].append(e.position) return {name:sorted(points) for name,points in result.items()} old_nets,new_nets=nets(before),nets(after) summary={status:sum(c['status']==status for c in changes) for status in ('added','removed','modified','moved')} return {'schema_version':'1.0','document_type':'schematic_comparison','summary':summary,'total_changed':len(changes), 'old':{'filename':old.filename,'compatibility':old.metadata.get('compatibility'),'sha256':old.metadata.get('sha256'),'sheet':old.sheet.bounds.model_dump(),'bounds':old.bounds.model_dump(),'diagnostics':[d.model_dump() for d in old.diagnostics]}, 'new':{'filename':new.filename,'compatibility':new.metadata.get('compatibility'),'sha256':new.metadata.get('sha256'),'sheet':new.sheet.bounds.model_dump(),'bounds':new.bounds.model_dump(),'diagnostics':[d.model_dump() for d in new.diagnostics]}, 'changes':changes,'nets':{'added':sorted(new_nets.keys()-old_nets.keys()),'removed':sorted(old_nets.keys()-new_nets.keys()), 'changed':sorted(name for name in old_nets.keys()&new_nets.keys() if old_nets[name]!=new_nets[name])}, 'limitations':['This is a structural/connection-graphics comparison, not a compiled electrical netlist.', 'Ambiguous identities remain added/removed. Geometry/designator fallback matches are heuristic.', 'Unsupported native objects retain the parser compatibility limits.'], 'matching':{'matched':len(pairs),'unmatched_old':len(removed),'unmatched_new':len(added),'ambiguous_groups':ambiguous}}