180 خطوط
10 KiB
Python
180 خطوط
10 KiB
Python
"""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}}
|