first commit where the whole thing works in at least common scenarios

This commit is contained in:
pszsh 2026-02-08 01:14:21 -08:00
parent 895139e14e
commit f64439f789
3 changed files with 320 additions and 180 deletions

View File

@ -10,8 +10,8 @@ class Layer:
self.name = name
self.manager = manager
self.visible = True
self.on_board_items = [] # List of {'uuid': str, 'class': str}
self.stored_items = [] # List of filenames (str) pointing to item JSONs
self.on_board_items = []
self.stored_items = []
def add_item(self, item):
if not item: return False
@ -24,37 +24,58 @@ class Layer:
self.on_board_items.append({'uuid': uuid, 'class': cls})
return True
def _safe_deselect(self, item):
if not item: return
try:
if hasattr(item, 'ClearSelected'): item.ClearSelected()
if hasattr(item, 'ClearBrightened'): item.ClearBrightened()
group = item.GetParentGroup()
if group:
if hasattr(group, 'ClearSelected'): group.ClearSelected()
if hasattr(group, 'ClearBrightened'): group.ClearBrightened()
except: pass
def _clear_selection_safe(self):
try:
sel = pcbnew.GetCurrentSelection()
items = [i for i in sel]
for item in items:
if hasattr(item, 'ClearSelected'): item.ClearSelected()
if hasattr(item, 'ClearBrightened'): item.ClearBrightened()
self._safe_deselect(item)
except: pass
def hide(self):
self.manager.log(f"Layer {self.name}: Hiding...")
pcbnew.Refresh() # Brush teeth (Start)
self.manager.log(f"Layer {self.name}: Hiding... ({len(self.on_board_items)} items)")
try:
board = pcbnew.GetBoard()
# 1. Deselect specific items
for entry in self.on_board_items:
try:
item = find_item_by_uuid(board, entry['uuid'])
if item: self._safe_deselect(item)
except: pass
# 2. Clear general selection
self._clear_selection_safe()
pcbnew.Refresh()
new_stored_items = list(self.stored_items)
items_to_remove = []
failed_entries = []
# 3. Serialize
for entry in self.on_board_items:
try:
uuid = entry['uuid']
self.manager.log(f" Processing {uuid}...")
item = find_item_by_uuid(board, uuid)
if not item:
self.manager.log(f"Hide: Item {uuid} not found (skipping)")
self.manager.log(f" Item not found on board (skipping)")
continue
sexpr, err = Serializer.serialize(item)
if not sexpr:
self.manager.log(f"Hide: Serialize failed {uuid}: {err}")
self.manager.log(f" Serialize failed: {err}")
failed_entries.append(entry)
continue
@ -64,32 +85,42 @@ class Layer:
filename = self.manager.storage.save_item(item_data, uuid)
if filename:
self.manager.log(f" Saved to {filename}")
new_stored_items.append(filename)
items_to_remove.append(item)
else:
self.manager.log(f" Storage save failed")
failed_entries.append(entry)
except Exception as e:
self.manager.log(f"Hide: Error entry {entry}: {e}")
self.manager.log(f" Error processing entry: {e}")
failed_entries.append(entry)
# 4. Remove
self.manager.log(f" Removing {len(items_to_remove)} items from board...")
for item in items_to_remove:
try:
if item.IsLocked(): item.SetLocked(False)
board.Remove(item)
except: pass
except Exception as e:
self.manager.log(f" Remove failed: {e}")
# 5. Update State
self.stored_items = new_stored_items
self.on_board_items = failed_entries
if not self.on_board_items: self.visible = False
self.manager.save()
self.manager.log("Hide: Complete")
self.manager.log(f"Hide: Complete. Stored: {len(self.stored_items)}, Remaining on board: {len(self.on_board_items)}")
except Exception as e:
self.manager.log(f"Hide: Critical Error: {e}")
self.manager.log(traceback.format_exc())
finally:
pcbnew.Refresh() # Brush teeth (End)
def show(self):
pcbnew.Refresh() # Brush teeth (Start)
try:
self.manager.log(f"Layer {self.name}: Showing (Force Restore)...")
self.manager.log(f"Layer {self.name}: Showing... Stored items: {len(self.stored_items)}")
board = pcbnew.GetBoard()
self.manager.is_restoring = True
@ -97,70 +128,93 @@ class Layer:
restored_files = []
for filename in self.stored_items:
self.manager.log(f" Processing file {filename}...")
try:
item_data = self.manager.storage.load_item(filename)
if not item_data: continue
uuid = item_data.get('uuid')
if not item_data:
self.manager.log(f" Failed to load JSON data")
continue
# Force Restore: Remove existing ghost if present
existing = find_item_by_uuid(board, uuid)
if existing:
try:
if existing.IsLocked(): existing.SetLocked(False)
board.Remove(existing)
except: pass
uuid = item_data.get('uuid')
self.manager.log(f" UUID: {uuid}")
# Note: Ghost detection removed to prevent segfaults on stale pointers.
# We assume items are not on board.
item, err = Serializer.deserialize(item_data, board)
if item:
try:
board.Add(item)
self.manager.log(f" Board.Add success")
new_uuid = get_item_uuid(item)
if new_uuid:
found = any(e['uuid'] == new_uuid for e in new_on_board_items)
if not found:
new_on_board_items.append({'uuid': new_uuid, 'class': item.GetClass()})
restored_files.append(filename)
else:
self.manager.log(f" Warning: Added item has no UUID")
except Exception as e:
self.manager.log(f"Show: Add failed {uuid}: {e}")
self.manager.log(f" Board.Add failed: {e}")
else:
self.manager.log(f" Deserialize failed: {err}")
except Exception as e:
self.manager.log(f"Show: Error file {filename}: {e}")
self.manager.log(f" Error processing file: {e}")
self.manager.log(traceback.format_exc())
self.on_board_items = new_on_board_items
for f in restored_files:
if f in self.stored_items: self.stored_items.remove(f)
# Rebuild stored_items to only contain those that failed to restore
failed_files = [f for f in self.stored_items if f not in restored_files]
self.stored_items = failed_files
self.visible = True
self.manager.is_restoring = False
self.manager.save()
self.manager.log("Show: Complete")
self.manager.log(f"Show: Complete. Restored: {len(restored_files)}, Failed/Remaining: {len(self.stored_items)}")
except Exception as e:
self.manager.log(f"Show: Critical Error: {e}")
self.manager.log(traceback.format_exc())
self.manager.is_restoring = False
finally:
pcbnew.Refresh() # Brush teeth (End)
def delete_content(self, delete_items_from_board):
board = pcbnew.GetBoard()
if self.visible:
if delete_items_from_board:
self._clear_selection_safe()
for entry in self.on_board_items:
item = find_item_by_uuid(board, entry['uuid'])
if item:
try:
if item.IsLocked(): item.SetLocked(False)
board.Remove(item)
pcbnew.Refresh() # Brush teeth (Start)
try:
board = pcbnew.GetBoard()
if self.visible:
if delete_items_from_board:
for entry in self.on_board_items:
try:
item = find_item_by_uuid(board, entry['uuid'])
if item: self._safe_deselect(item)
except: pass
self.on_board_items = []
else:
if not delete_items_from_board:
self.manager.is_restoring = True
for filename in self.stored_items:
item_data = self.manager.storage.load_item(filename)
if item_data:
item, err = Serializer.deserialize(item_data, board)
if item: board.Add(item)
self.manager.is_restoring = False
self.stored_items = []
self.manager.save()
self._clear_selection_safe()
for entry in self.on_board_items:
item = find_item_by_uuid(board, entry['uuid'])
if item:
try:
if item.IsLocked(): item.SetLocked(False)
board.Remove(item)
except: pass
self.on_board_items = []
else:
if not delete_items_from_board:
self.manager.is_restoring = True
for filename in self.stored_items:
item_data = self.manager.storage.load_item(filename)
if item_data:
item, err = Serializer.deserialize(item_data, board)
if item: board.Add(item)
self.manager.is_restoring = False
self.stored_items = []
self.manager.save()
finally:
pcbnew.Refresh() # Brush teeth (End)
def clear_items(self):
self.delete_content(delete_items_from_board=True)
@ -180,19 +234,23 @@ class Layer:
return l
def add_from_text(self, text):
uuids = set(re.findall(r'\(uuid\s+"([^"]+)"\)', text))
if not uuids: return 0, "No UUIDs found."
board = pcbnew.GetBoard()
count = 0
for uuid in uuids:
try:
item = board.GetItem(pcbnew.KIID(uuid))
if item:
trackable = resolve_to_trackable_item(item)
if trackable and self.add_item(trackable): count += 1
except: pass
self.manager.save()
return count, f"Added {count} items."
pcbnew.Refresh() # Brush teeth (Start)
try:
uuids = set(re.findall(r'\(uuid\s+"([^"]+)"\)', text))
if not uuids: return 0, "No UUIDs found."
board = pcbnew.GetBoard()
count = 0
for uuid in uuids:
try:
item = board.GetItem(pcbnew.KIID(uuid))
if item:
trackable = resolve_to_trackable_item(item)
if trackable and self.add_item(trackable): count += 1
except: pass
self.manager.save()
return count, f"Added {count} items."
finally:
pcbnew.Refresh() # Brush teeth (End)
def _get_selection_candidates(self, board):
candidates = []
@ -202,7 +260,6 @@ class Layer:
except: pass
if not candidates:
# Fallback scan
for lst in [board.GetTracks(), board.GetFootprints(), board.GetDrawings(), board.Zones(), board.Groups()]:
try:
for item in lst:
@ -211,54 +268,62 @@ class Layer:
return candidates
def add_selection(self):
board = pcbnew.GetBoard()
candidates = self._get_selection_candidates(board)
items_to_add = {}
def process_item(item):
if item.GetClass() == "PCB_GROUP":
try:
group = pcbnew.Cast_to_PCB_GROUP(item)
for sub in group.GetItems(): process_item(sub)
except: pass
else:
trackable = resolve_to_trackable_item(item)
if trackable:
uuid = get_item_uuid(trackable)
if uuid: items_to_add[uuid] = trackable
pcbnew.Refresh() # Brush teeth (Start)
try:
board = pcbnew.GetBoard()
candidates = self._get_selection_candidates(board)
items_to_add = {}
def process_item(item):
if item.GetClass() == "PCB_GROUP":
try:
group = pcbnew.Cast_to_PCB_GROUP(item)
for sub in group.GetItems(): process_item(sub)
except: pass
else:
trackable = resolve_to_trackable_item(item)
if trackable:
uuid = get_item_uuid(trackable)
if uuid: items_to_add[uuid] = trackable
for item in candidates: process_item(item)
count = 0
for uuid, item in items_to_add.items():
if self.add_item(item): count += 1
self.manager.save()
return count, f"Added {count} items."
for item in candidates: process_item(item)
count = 0
for uuid, item in items_to_add.items():
if self.add_item(item): count += 1
self.manager.save()
return count, f"Added {count} items."
finally:
pcbnew.Refresh() # Brush teeth (End)
def remove_selection(self):
board = pcbnew.GetBoard()
candidates = self._get_selection_candidates(board)
uuids_to_remove = set()
def collect_uuids(item):
if item.GetClass() == "PCB_GROUP":
try:
group = pcbnew.Cast_to_PCB_GROUP(item)
for sub in group.GetItems(): collect_uuids(sub)
except: pass
else:
trackable = resolve_to_trackable_item(item)
if trackable:
uuid = get_item_uuid(trackable)
if uuid: uuids_to_remove.add(uuid)
pcbnew.Refresh() # Brush teeth (Start)
try:
board = pcbnew.GetBoard()
candidates = self._get_selection_candidates(board)
uuids_to_remove = set()
def collect_uuids(item):
if item.GetClass() == "PCB_GROUP":
try:
group = pcbnew.Cast_to_PCB_GROUP(item)
for sub in group.GetItems(): collect_uuids(sub)
except: pass
else:
trackable = resolve_to_trackable_item(item)
if trackable:
uuid = get_item_uuid(trackable)
if uuid: uuids_to_remove.add(uuid)
for item in candidates: collect_uuids(item)
initial_count = len(self.on_board_items)
self.on_board_items = [e for e in self.on_board_items if e['uuid'] not in uuids_to_remove]
removed = initial_count - len(self.on_board_items)
self.manager.save()
return removed, f"Removed {removed} items."
for item in candidates: collect_uuids(item)
initial_count = len(self.on_board_items)
self.on_board_items = [e for e in self.on_board_items if e['uuid'] not in uuids_to_remove]
removed = initial_count - len(self.on_board_items)
self.manager.save()
return removed, f"Removed {removed} items."
finally:
pcbnew.Refresh() # Brush teeth (End)
def inspect(self):
return f"Layer: {self.name}\nVisible: {self.visible}\nOn Board: {len(self.on_board_items)}\nStored: {len(self.stored_items)}"

View File

@ -13,14 +13,37 @@ class LayerListener(pcbnew.BOARD_LISTENER):
self.mgr = mgr
def OnBoardItemAdded(self, item):
if self.mgr.is_restoring: return
if self.mgr.active_layer_index is not None:
try:
layer = self.mgr.layers[self.mgr.active_layer_index]
if layer.visible:
trackable = resolve_to_trackable_item(item)
if trackable: layer.add_item(trackable)
except: pass
# 1. Record State (The "Wake Up Tape")
# Capture UUID before any refresh invalidates the item pointer or tool state.
target_uuid = get_item_uuid(item)
if not target_uuid: return
# 2. Brush Teeth (Start) - Sync with reality
# This ensures we are facing the actual playing field.
pcbnew.Refresh()
try:
if self.mgr.is_restoring: return
# 3. Re-acquire State (Watch the tape)
# The 'item' pointer passed by the event might be unsafe now.
# We look up the item freshly from the board using the recorded UUID.
board = pcbnew.GetBoard()
fresh_item = find_item_by_uuid(board, target_uuid)
if not fresh_item: return
# 4. Perform Action
if self.mgr.active_layer_index is not None:
try:
layer = self.mgr.layers[self.mgr.active_layer_index]
if layer.visible:
trackable = resolve_to_trackable_item(fresh_item)
if trackable: layer.add_item(trackable)
except: pass
finally:
# 5. Brush Teeth (End) - Finalize
pcbnew.Refresh()
class LayerManager:
def __init__(self):
@ -108,74 +131,86 @@ class LayerManager:
return False
def analyze_clipboard_text(self, text):
self.log(f"Analyze: Input text length: {len(text)}")
uuids = set(re.findall(r'\(uuid\s+"([^"]+)"\)', text))
if not uuids: return self.get_log_text()
board = pcbnew.GetBoard()
matched = 0
def check(item):
try:
if get_item_uuid(item) in uuids: return 1
except: pass
return 0
pcbnew.Refresh() # Start
try:
for lst in [board.GetTracks(), board.GetFootprints(), board.GetDrawings(), board.Zones()]:
for item in lst: matched += check(item)
except: pass
self.log(f"Analyze: Total Matches on Board: {matched}")
return self.get_log_text()
self.log(f"Analyze: Input text length: {len(text)}")
uuids = set(re.findall(r'\(uuid\s+"([^"]+)"\)', text))
if not uuids: return self.get_log_text()
board = pcbnew.GetBoard()
matched = 0
def check(item):
try:
if get_item_uuid(item) in uuids: return 1
except: pass
return 0
try:
for lst in [board.GetTracks(), board.GetFootprints(), board.GetDrawings(), board.Zones()]:
for item in lst: matched += check(item)
except: pass
self.log(f"Analyze: Total Matches on Board: {matched}")
return self.get_log_text()
finally:
pcbnew.Refresh() # End
def recover_orphans(self):
if not self.storage.setup_paths(): return 0, "Storage not ready"
tracked_uuids = set()
stored_files = set()
board = pcbnew.GetBoard()
for l in self.layers:
for f in l.stored_items: stored_files.add(f)
for entry in l.on_board_items: tracked_uuids.add(entry['uuid'])
orphans = []
pcbnew.Refresh() # Start
try:
for f in os.listdir(self.storage.session_dir):
if f.endswith(".json") and f != "$Master$.json" and f not in stored_files:
data = self.storage.load_item(f)
if not data or 'uuid' not in data: continue
uuid = data['uuid']
if uuid in tracked_uuids or find_item_by_uuid(board, uuid): continue
orphans.append(f)
except Exception as e: return 0, f"Error scanning dir: {e}"
if not orphans: return 0, "No orphans found"
rec_layer = self.create_layer("Recovered")
rec_layer.stored_items = orphans
rec_layer.visible = False
self.save()
return len(orphans), f"Recovered {len(orphans)} items."
if not self.storage.setup_paths(): return 0, "Storage not ready"
tracked_uuids = set()
stored_files = set()
board = pcbnew.GetBoard()
for l in self.layers:
for f in l.stored_items: stored_files.add(f)
for entry in l.on_board_items: tracked_uuids.add(entry['uuid'])
orphans = []
try:
for f in os.listdir(self.storage.session_dir):
if f.endswith(".json") and f != "$Master$.json" and f not in stored_files:
data = self.storage.load_item(f)
if not data or 'uuid' not in data: continue
uuid = data['uuid']
if uuid in tracked_uuids or find_item_by_uuid(board, uuid): continue
orphans.append(f)
except Exception as e: return 0, f"Error scanning dir: {e}"
if not orphans: return 0, "No orphans found"
rec_layer = self.create_layer("Recovered")
rec_layer.stored_items = orphans
rec_layer.visible = False
self.save()
return len(orphans), f"Recovered {len(orphans)} items."
finally:
pcbnew.Refresh() # End
def hard_refresh(self):
self.log("Hard Refresh: Starting...")
board = pcbnew.GetBoard()
for layer in self.layers:
if layer.visible:
valid_on_board = []
for entry in layer.on_board_items:
uuid = entry['uuid']
item = find_item_by_uuid(board, uuid)
if item:
sexpr, err = Serializer.serialize(item)
if sexpr: valid_on_board.append(entry)
else:
self.log(f"Refresh: Item {uuid} missing from visible layer {layer.name}")
layer.on_board_items = valid_on_board
else:
for filename in layer.stored_items:
data = self.storage.load_item(filename)
if data and 'uuid' in data:
uuid = data['uuid']
pcbnew.Refresh() # Start
try:
self.log("Hard Refresh: Starting...")
board = pcbnew.GetBoard()
for layer in self.layers:
if layer.visible:
valid_on_board = []
for entry in layer.on_board_items:
uuid = entry['uuid']
item = find_item_by_uuid(board, uuid)
if item:
self.log(f"Refresh: Found ghost {uuid} from hidden layer {layer.name}. Removing.")
try: board.Remove(item)
except: pass
self.save()
self.log("Hard Refresh: Complete")
sexpr, err = Serializer.serialize(item)
if sexpr: valid_on_board.append(entry)
else:
self.log(f"Refresh: Item {uuid} missing from visible layer {layer.name}")
layer.on_board_items = valid_on_board
else:
for filename in layer.stored_items:
data = self.storage.load_item(filename)
if data and 'uuid' in data:
uuid = data['uuid']
item = find_item_by_uuid(board, uuid)
if item:
self.log(f"Refresh: Found ghost {uuid} from hidden layer {layer.name}. Removing.")
try: board.Remove(item)
except: pass
self.save()
self.log("Hard Refresh: Complete")
finally:
pcbnew.Refresh() # End
manager = LayerManager()

View File

@ -1,5 +1,6 @@
# backend/serializer.py
import pcbnew
import traceback
class Serializer:
@staticmethod
@ -64,17 +65,56 @@ class Serializer:
except Exception as e:
return None, f"Constructor failed for {cls_name}: {e}"
# Deserialize with KiCad version compatibility
# KiCad 9+ requires (data, source_name), older versions take (data)
# Ensure sexpr is a string
if not isinstance(sexpr, str):
sexpr = str(sexpr)
# DEBUG LOGGING
print(f"[DEBUG] Deserializing {cls_name}")
print(f"[DEBUG] Sexpr (first 50 chars): {sexpr[:50]}")
lr = None
try:
# Primary method: KiCad 9+
lr = pcbnew.STRING_LINE_READER(sexpr, "restore")
except:
except TypeError:
# Fallback: Older KiCad versions (1 arg)
try:
lr = pcbnew.STRING_LINE_READER(sexpr)
except Exception as e:
return None, f"STRING_LINE_READER init failed: {e}"
return None, f"STRING_LINE_READER init failed (Fallback): {e}"
except Exception as e:
return None, f"STRING_LINE_READER init failed: {e}"
if lr is None:
return None, "STRING_LINE_READER is None"
print(f"[DEBUG] LineReader created: {lr}")
try:
item.Deserialize(lr)
except Exception as e:
print(f"[DEBUG] Deserialize Exception: {e}")
item.Deserialize(lr)
# Check for the specific Protobuf overload mismatch
if "google::protobuf::Any" in str(e):
print("[DEBUG] Protobuf overload mismatch detected. Attempting PCB_IO_KICAD_SEXPR Parse fallback...")
try:
# Hail Mary: Try using the IO plugin directly to parse the string
# This bypasses the ambiguous Deserialize overload on the item itself
plugin = pcbnew.PCB_IO_KICAD_SEXPR()
if hasattr(plugin, 'Parse'):
# Parse usually returns a new item
parsed_item = plugin.Parse(sexpr)
if parsed_item:
print("[DEBUG] Fallback Parse successful!")
return parsed_item, None
except Exception as ex:
print(f"[DEBUG] Fallback Parse failed: {ex}")
# If fallback didn't work or wasn't applicable, re-raise
raise e
return item, None
except Exception as e:
return None, f"Deserialization error for {cls_name}: {e}"