ArdaandClaude Opus 5.5
08b4dec532
Right-click a component (or the app) with two or more children to move them up and down; OK applies the new order, which is the order they run in. On a device the order goes there first via Sox reorder, checked against the device's children, and the file is left alone if it is refused. Undoable. Also: device host in editor.properties. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1706 lines
63 KiB
Python
1706 lines
63 KiB
Python
"""Opening, saving and structural editing of a Sedona app file, plus the navigation tree."""
|
|
|
|
import copy
|
|
from datetime import datetime
|
|
import os
|
|
from tkinter import filedialog, messagebox
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import app_state
|
|
import live_hooks
|
|
import palette
|
|
import properties
|
|
import schema
|
|
import undo
|
|
import wiresheet
|
|
from sax_dom import (
|
|
find_component_by_path,
|
|
find_parent_element,
|
|
get_component_path,
|
|
read_meta_property,
|
|
restore_children,
|
|
snapshot_children,
|
|
update_meta_property,
|
|
)
|
|
|
|
|
|
_tree_filled = None
|
|
|
|
|
|
def set_tree_filled_callback(callback):
|
|
"""Registers what should run once the navigation tree has been rebuilt."""
|
|
global _tree_filled
|
|
_tree_filled = callback
|
|
|
|
|
|
_disconnect_device = None
|
|
|
|
|
|
def set_disconnect_device_callback(callback):
|
|
"""Registers how to drop the live device session before a file takes its place."""
|
|
global _disconnect_device
|
|
_disconnect_device = callback
|
|
|
|
|
|
def parse_xml_to_tree(treeview, parent_node, xml_element):
|
|
"""Recursively traverses XML elements and adds <comp> tags to the Treeview."""
|
|
for child in xml_element:
|
|
if child.tag == "comp":
|
|
comp_name = child.get("name", "Unknown")
|
|
comp_id = child.get("id")
|
|
|
|
if comp_id is not None:
|
|
display_text = f"{comp_name} [id: {comp_id}]"
|
|
else:
|
|
display_text = comp_name
|
|
|
|
node_id = treeview.insert(
|
|
parent_node, "end", text=display_text, open=False
|
|
)
|
|
app_state.tree_element_map[node_id] = child
|
|
parse_xml_to_tree(treeview, node_id, child)
|
|
|
|
|
|
def _expansion_key(treeview, item, parent_key):
|
|
"""Names a tree row in a way that survives a rebuild of the tree.
|
|
|
|
A component's id is stable across edits and across a fresh DOM read from the
|
|
device; a row without one (the root, or an id-less file component) falls back to
|
|
its name under its parent's key.
|
|
"""
|
|
elem = app_state.tree_element_map.get(item)
|
|
comp_id = elem.get("id") if elem is not None else None
|
|
if comp_id is not None and treeview.parent(item) != "":
|
|
return ("id", comp_id)
|
|
name = elem.get("name", "") if elem is not None else treeview.item(item, "text")
|
|
return ("path", parent_key, name)
|
|
|
|
|
|
def _collect_open_keys(treeview, parent="", parent_key=None, found=None):
|
|
"""Returns the keys of every row that is expanded right now."""
|
|
if found is None:
|
|
found = set()
|
|
for item in treeview.get_children(parent):
|
|
key = _expansion_key(treeview, item, parent_key)
|
|
if treeview.item(item, "open"):
|
|
found.add(key)
|
|
_collect_open_keys(treeview, item, key, found)
|
|
return found
|
|
|
|
|
|
def _reopen_keys(treeview, open_keys, parent="", parent_key=None):
|
|
"""Expands again the rows that were expanded before the rebuild.
|
|
|
|
Setting the open flag in code does not raise <<TreeviewOpen>>, so this never
|
|
starts a live child load of its own.
|
|
"""
|
|
for item in treeview.get_children(parent):
|
|
key = _expansion_key(treeview, item, parent_key)
|
|
if key in open_keys:
|
|
treeview.item(item, open=True)
|
|
_reopen_keys(treeview, open_keys, item, key)
|
|
|
|
|
|
def refresh_treeview_from_dom(xml_root):
|
|
"""Rebuilds navigation tree structures using current XML DOM trees.
|
|
|
|
Rows the user had expanded stay expanded; only the user's own "-" closes one.
|
|
"""
|
|
tree = app_state.tree
|
|
open_keys = _collect_open_keys(tree)
|
|
app_state.tree_element_map.clear()
|
|
|
|
for item in tree.get_children():
|
|
tree.delete(item)
|
|
|
|
if not app_state.current_file_path or xml_root is None:
|
|
if _tree_filled is not None:
|
|
_tree_filled()
|
|
return
|
|
|
|
if app_state.live_mode:
|
|
source = f"{app_state.device_host}:{app_state.device_port}"
|
|
else:
|
|
source = os.path.basename(app_state.current_file_path)
|
|
# The row is the <app> component itself; where it came from goes in brackets.
|
|
# Inserted plain. refresh_dirty_display puts the unsaved tag on or off, and runs
|
|
# after every rebuild, so a fresh row picks up the right state there.
|
|
root_node = tree.insert("", "end", text=f"app ({source})", open=True)
|
|
|
|
app_element = xml_root.find("app")
|
|
if app_element is not None:
|
|
app_state.tree_element_map[root_node] = app_element
|
|
parse_xml_to_tree(tree, root_node, app_element)
|
|
else:
|
|
parse_xml_to_tree(tree, root_node, xml_root)
|
|
|
|
_reopen_keys(tree, open_keys)
|
|
|
|
if _tree_filled is not None:
|
|
_tree_filled()
|
|
|
|
|
|
def reset_to_startup_state():
|
|
"""Returns the app to its just-launched state: no file, empty tree, empty sheet.
|
|
|
|
A load that fails part way has already cleared the previous file, so leaving the
|
|
half-built state on screen would show a tree and a DOM that no longer agree. This
|
|
is the one place that unwinds all of it.
|
|
"""
|
|
app_state.current_file_path = None
|
|
app_state.xml_root_element = None
|
|
app_state.current_sheet_parent = None
|
|
app_state.selected_canvas_tag = None
|
|
app_state.selected_canvas_tags.clear()
|
|
app_state.selected_tree_item_id = None
|
|
app_state.link_mark_element = None
|
|
app_state.selected_link_tag = None
|
|
app_state.selected_link_tags.clear()
|
|
|
|
app_state.active_components_list.clear()
|
|
app_state.tree_element_map.clear()
|
|
app_state.canvas_comp_map.clear()
|
|
app_state.canvas_tag_by_element.clear()
|
|
app_state.component_slot_offsets.clear()
|
|
app_state.canvas_link_map.clear()
|
|
app_state.final_components_registry.clear()
|
|
app_state.active_schema_kits.clear()
|
|
app_state.pending_schema_kits.clear()
|
|
|
|
undo.reset()
|
|
app_state.mark_clean()
|
|
app_state.clear_status()
|
|
|
|
if app_state.tree is not None:
|
|
for item in app_state.tree.get_children():
|
|
app_state.tree.delete(item)
|
|
|
|
if _tree_filled is not None:
|
|
_tree_filled()
|
|
|
|
if app_state.canvas is not None:
|
|
app_state.canvas.delete("all")
|
|
wiresheet.draw_grid()
|
|
|
|
if app_state.window is not None:
|
|
app_state.window.title(app_state.base_window_title())
|
|
|
|
|
|
def open_file():
|
|
if not app_state.SEDONA_HOME:
|
|
messagebox.showwarning("Action Blocked", "Cannot load SAX profiles because 'Sedona home folder' configuration is missing or empty.")
|
|
return
|
|
|
|
file_path = filedialog.askopenfilename(
|
|
title="Select SAX File",
|
|
filetypes=[("SAX Files", "*.sax"), ("XML Files", "*.xml"), ("All Files", "*.*")],
|
|
)
|
|
if not file_path:
|
|
return
|
|
|
|
# A session is a file or a device, never both. The device goes only once a file has
|
|
# actually been chosen, so cancelling the dialog leaves the connection as it was.
|
|
if app_state.device_session is not None and _disconnect_device is not None:
|
|
_disconnect_device()
|
|
|
|
canvas = app_state.canvas
|
|
tree = app_state.tree
|
|
|
|
try:
|
|
app_state.current_file_path = file_path
|
|
app_state.active_components_list.clear()
|
|
app_state.pending_schema_kits.clear()
|
|
app_state.clear_status()
|
|
app_state.selected_canvas_tag = None
|
|
app_state.selected_canvas_tags.clear()
|
|
app_state.current_sheet_parent = None
|
|
undo.reset()
|
|
app_state.mark_clean()
|
|
|
|
app_state.tree_element_map.clear()
|
|
app_state.canvas_comp_map.clear()
|
|
app_state.canvas_tag_by_element.clear()
|
|
for item in tree.get_children():
|
|
tree.delete(item)
|
|
canvas.delete("all")
|
|
wiresheet.draw_grid()
|
|
|
|
xml_tree = ET.parse(file_path)
|
|
app_state.xml_root_element = xml_tree.getroot()
|
|
|
|
schema.load_schema_kits_and_manifests()
|
|
|
|
file_name = os.path.basename(file_path)
|
|
root_node = tree.insert("", "end", text=file_name, open=True)
|
|
|
|
app_element = app_state.xml_root_element.find("app")
|
|
if app_element is not None:
|
|
parse_xml_to_tree(tree, root_node, app_element)
|
|
else:
|
|
parse_xml_to_tree(tree, root_node, app_state.xml_root_element)
|
|
|
|
if _tree_filled is not None:
|
|
_tree_filled()
|
|
|
|
except schema.MissingManifestError as missing:
|
|
messagebox.showerror(
|
|
"Kit manifest missing",
|
|
f"'{os.path.basename(file_path)}' was not opened.\n\n"
|
|
+ schema.describe_missing_manifests(missing)
|
|
+ "\n\nFix the Sedona home folder, then open the file again."
|
|
)
|
|
reset_to_startup_state()
|
|
except Exception as e:
|
|
messagebox.showerror("Error", f"Failed to parse the file:\n{str(e)}")
|
|
reset_to_startup_state()
|
|
|
|
|
|
def sync_live_metadata_to_dom():
|
|
"""Internal helper to clean up deleted elements and sync coordinates into the root XML object."""
|
|
if app_state.xml_root_element is None:
|
|
return
|
|
|
|
live_elements_meta = {}
|
|
for cached_tag, cached_element in app_state.canvas_comp_map.items():
|
|
comp_id = cached_element.get("id")
|
|
comp_name = cached_element.get("name")
|
|
meta_prop = cached_element.find("prop[@name='meta']")
|
|
if meta_prop is not None:
|
|
live_elements_meta[(comp_id, comp_name)] = meta_prop.get("val")
|
|
|
|
def sync_and_prune_tree(target_xml_parent):
|
|
children_to_examine = list(target_xml_parent)
|
|
for child in children_to_examine:
|
|
if child.tag == "comp":
|
|
comp_id = child.get("id")
|
|
comp_name = child.get("name")
|
|
|
|
is_element_alive = False
|
|
for c_elem in app_state.active_components_list:
|
|
if c_elem.get("id") == comp_id and c_elem.get("name") == comp_name:
|
|
is_element_alive = True
|
|
break
|
|
|
|
is_previously_rendered = any(
|
|
elem.get("id") == comp_id and elem.get("name") == comp_name
|
|
for elem in app_state.canvas_comp_map.values()
|
|
) or (comp_id, comp_name) in live_elements_meta
|
|
|
|
if is_previously_rendered and not is_element_alive:
|
|
target_xml_parent.remove(child)
|
|
continue
|
|
|
|
if (comp_id, comp_name) in live_elements_meta:
|
|
tgt_meta = child.find("prop[@name='meta']")
|
|
if tgt_meta is None:
|
|
tgt_meta = ET.SubElement(child, "prop", name="meta")
|
|
tgt_meta.set("val", live_elements_meta[(comp_id, comp_name)])
|
|
|
|
sync_and_prune_tree(child)
|
|
|
|
sync_and_prune_tree(app_state.xml_root_element)
|
|
|
|
|
|
def execute_direct_overwrite_save():
|
|
"""Overwrites the existing file directly without adding timestamp formats or prompts."""
|
|
if not app_state.current_file_path or app_state.xml_root_element is None:
|
|
return
|
|
|
|
try:
|
|
sync_live_metadata_to_dom()
|
|
|
|
if not reconcile_schema_kits_before_save():
|
|
return
|
|
|
|
with open(app_state.current_file_path, "wb") as f:
|
|
f.write(b'<?xml version="1.0" encoding="UTF-8"?>\n')
|
|
ET.ElementTree(app_state.xml_root_element).write(f, encoding="utf-8", xml_declaration=False)
|
|
|
|
app_state.mark_clean()
|
|
undo.mark_saved()
|
|
refresh_treeview_from_dom(app_state.xml_root_element)
|
|
app_state.set_status(f"Saved {os.path.basename(app_state.current_file_path)}.")
|
|
|
|
messagebox.showinfo("Success", f"Direct changes saved onto:\n{os.path.basename(app_state.current_file_path)}")
|
|
except Exception as e:
|
|
messagebox.showerror("Save Error", f"Failed to write direct update stream:\n{str(e)}")
|
|
|
|
|
|
def write_sax_to_path(path):
|
|
"""Writes the current DOM to a .sax file at the given path."""
|
|
if app_state.xml_root_element is None:
|
|
raise RuntimeError("No DOM to write")
|
|
sync_live_metadata_to_dom()
|
|
with open(path, "wb") as f:
|
|
f.write(b'<?xml version="1.0" encoding="UTF-8"?>\n')
|
|
ET.ElementTree(app_state.xml_root_element).write(
|
|
f, encoding="utf-8", xml_declaration=False
|
|
)
|
|
|
|
|
|
def save_file_as():
|
|
"""Saves the tracking tree document hierarchy targeting an explicit location choice dialog."""
|
|
if not app_state.current_file_path or app_state.xml_root_element is None:
|
|
messagebox.showinfo("Save Info", "No active profile has been opened yet.")
|
|
return
|
|
|
|
try:
|
|
sync_live_metadata_to_dom()
|
|
|
|
if app_state.ADD_TIMESTAMP:
|
|
base_dir = os.path.dirname(app_state.current_file_path)
|
|
orig_filename = os.path.basename(app_state.current_file_path)
|
|
name_part, ext_part = os.path.splitext(orig_filename)
|
|
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
suggested_name = f"{name_part}_{timestamp}{ext_part}"
|
|
|
|
target_path = filedialog.asksaveasfilename(
|
|
initialdir=base_dir,
|
|
initialfile=suggested_name,
|
|
title="Save Timestamps Layout As",
|
|
filetypes=[
|
|
("SAX Files", "*.sax"),
|
|
("XML Files", "*.xml"),
|
|
("All Files", "*.*"),
|
|
]
|
|
)
|
|
|
|
if not target_path:
|
|
return
|
|
else:
|
|
target_path = app_state.current_file_path
|
|
|
|
if not reconcile_schema_kits_before_save():
|
|
return
|
|
|
|
with open(target_path, "wb") as f:
|
|
f.write(b'<?xml version="1.0" encoding="UTF-8"?>\n')
|
|
ET.ElementTree(app_state.xml_root_element).write(f, encoding="utf-8", xml_declaration=False)
|
|
|
|
app_state.mark_clean()
|
|
undo.mark_saved()
|
|
|
|
refresh_treeview_from_dom(app_state.xml_root_element)
|
|
|
|
folder_name = os.path.basename(os.path.dirname(target_path))
|
|
file_name = os.path.basename(target_path)
|
|
app_state.set_status(f"Saved {file_name}.")
|
|
|
|
messagebox.showinfo("Success", f"File Saved\nFolder: {folder_name}\nFile: {file_name}")
|
|
|
|
except Exception as e:
|
|
messagebox.showerror("Save Error", f"An anomaly prevented saving the XML file stream:\n{str(e)}")
|
|
|
|
|
|
MAX_NAME_LENGTH = 7 # every component name in a Sedona app observes this
|
|
|
|
|
|
def check_component_name(name, parent=None, element=None):
|
|
"""Why `name` cannot name a component, or None if it can.
|
|
|
|
Mirrors Sedona's Component.checkName -- 1 to 7 ASCII characters, a letter first, then
|
|
letters, digits or '_' -- plus the uniqueness among siblings that OfflineApp enforces.
|
|
`element` is the component being renamed, which may keep its own name.
|
|
"""
|
|
if len(name) == 0:
|
|
return "The name is empty."
|
|
if len(name) > MAX_NAME_LENGTH:
|
|
return f"The name is longer than {MAX_NAME_LENGTH} characters."
|
|
for i, c in enumerate(name):
|
|
if "A" <= c <= "Z" or "a" <= c <= "z":
|
|
continue
|
|
if i == 0:
|
|
return "The name must start with a letter (A-Z or a-z)."
|
|
if "0" <= c <= "9" or c == "_":
|
|
continue
|
|
return f"'{c}' is not allowed; use only letters, digits and '_'."
|
|
if parent is not None:
|
|
for sibling in parent:
|
|
if sibling.tag == "comp" and sibling is not element and sibling.get("name") == name:
|
|
return f"'{name}' is already used by another component in the same parent."
|
|
return None
|
|
|
|
|
|
def rename_component(comp_element, new_name):
|
|
"""Renames a component and rewrites the link paths that ran through it.
|
|
|
|
Returns True when the rename happened. The name is validated here as well as in the
|
|
dialog, so nothing can put a name into the file that Sedona would refuse to load.
|
|
"""
|
|
root = app_state.xml_root_element
|
|
if root is None or comp_element.tag != "comp":
|
|
return False
|
|
|
|
old_name = comp_element.get("name", "")
|
|
if new_name == old_name:
|
|
return False
|
|
|
|
parent = find_parent_element(root, comp_element)
|
|
err = check_component_name(new_name, parent, comp_element)
|
|
if err is not None:
|
|
messagebox.showwarning("Rename Component", f"'{new_name}' is not a valid name.\n\n{err}")
|
|
return False
|
|
|
|
if app_state.live_mode:
|
|
if not live_hooks.live_rename_component(comp_element, new_name):
|
|
live_hooks.warn_failure("Rename Component",
|
|
f"{old_name} was not renamed on the device")
|
|
return False
|
|
|
|
old_path = get_component_path(comp_element, root)
|
|
new_path = old_path[:-len(old_name)] + new_name
|
|
|
|
# Links address components by path, so every one through the old name moves with it.
|
|
link_changes = []
|
|
links_elem = root.find("links")
|
|
if links_elem is not None:
|
|
for link in links_elem:
|
|
if link.tag != "link":
|
|
continue
|
|
for attr in ("from", "to"):
|
|
ref = link.get(attr, "")
|
|
if ref.startswith(old_path + ".") or ref.startswith(old_path + "/"):
|
|
link_changes.append((link, attr, ref, new_path + ref[len(old_path):]))
|
|
|
|
comp_element.set("name", new_name)
|
|
for link, attr, _, new_ref in link_changes:
|
|
link.set(attr, new_ref)
|
|
|
|
command_class = (undo.LiveRenameCommand if app_state.live_mode
|
|
else undo.RenameComponentCommand)
|
|
undo.push(command_class(
|
|
comp_element, old_name, new_name, link_changes,
|
|
label=f"rename {old_name} to {new_name}"
|
|
))
|
|
|
|
app_state.mark_dirty()
|
|
refresh_treeview_from_dom(root)
|
|
wiresheet.render_current_sheet()
|
|
|
|
where = " on the device" if app_state.live_mode else ""
|
|
app_state.set_status(f"{old_name} renamed to {new_name}{where}.")
|
|
return True
|
|
|
|
|
|
def reorder_children(parent, new_order):
|
|
"""Gives `parent`'s child components a new order, which is the order they run in.
|
|
|
|
Returns True when the order changed. On a device the new order goes there first,
|
|
and the file is left alone if it does not.
|
|
"""
|
|
root = app_state.xml_root_element
|
|
if root is None:
|
|
return False
|
|
old_order = [child for child in parent if child.tag == "comp"]
|
|
new_order = list(new_order)
|
|
if new_order == old_order:
|
|
return False
|
|
if (len(new_order) != len(old_order)
|
|
or {id(c) for c in new_order} != {id(c) for c in old_order}):
|
|
return False
|
|
|
|
name = parent.get("name", "app")
|
|
if app_state.live_mode:
|
|
if not live_hooks.live_reorder_children(parent, new_order):
|
|
live_hooks.warn_failure("Reorder", f"{name} was not reordered on the device")
|
|
return False
|
|
|
|
undo.apply_child_order(parent, new_order)
|
|
command_class = (undo.LiveReorderChildrenCommand if app_state.live_mode
|
|
else undo.ReorderChildrenCommand)
|
|
undo.push(command_class(parent, old_order, new_order, label=f"reorder {name}"))
|
|
|
|
app_state.mark_dirty()
|
|
refresh_treeview_from_dom(root)
|
|
wiresheet.render_current_sheet()
|
|
|
|
where = " on the device" if app_state.live_mode else ""
|
|
app_state.set_status(f"{name} reordered{where}.")
|
|
return True
|
|
|
|
|
|
def allocate_component_id():
|
|
"""Lowest unused id, matching how this app's ids are dense with no gaps."""
|
|
used = set()
|
|
if app_state.xml_root_element is not None:
|
|
for element in app_state.xml_root_element.iter("comp"):
|
|
try:
|
|
used.add(int(element.get("id")))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
candidate = 1
|
|
while candidate in used:
|
|
candidate += 1
|
|
return candidate
|
|
|
|
|
|
def unique_child_name(parent, base_name):
|
|
"""Trims to the 7-character limit and disambiguates against existing siblings."""
|
|
existing = {child.get("name") for child in parent if child.tag == "comp"}
|
|
|
|
candidate = base_name[:MAX_NAME_LENGTH]
|
|
if candidate and candidate not in existing:
|
|
return candidate
|
|
|
|
for counter in range(1, 10000):
|
|
suffix = str(counter)
|
|
stem = base_name[:max(1, MAX_NAME_LENGTH - len(suffix))]
|
|
candidate = f"{stem}{suffix}"[:MAX_NAME_LENGTH]
|
|
if candidate not in existing:
|
|
return candidate
|
|
|
|
return base_name[:MAX_NAME_LENGTH]
|
|
|
|
|
|
def duplicate_name(name, taken):
|
|
"""A name for a copy of `name`: its stem with the next free number, Add1 -> Add2.
|
|
|
|
Counting on from a trailing number, rather than appending another, keeps a run of
|
|
copies reading Add1, Add2, Add3 instead of Add1, Add11, Add111.
|
|
"""
|
|
stem = name.rstrip("0123456789") or "c"
|
|
digits = name[len(stem):]
|
|
counter = int(digits) + 1 if digits else 1
|
|
while True:
|
|
suffix = str(counter)
|
|
candidate = stem[:max(1, MAX_NAME_LENGTH - len(suffix))] + suffix
|
|
if candidate not in taken:
|
|
return candidate
|
|
counter += 1
|
|
|
|
|
|
def current_sheet_container():
|
|
"""The element a newly placed component belongs to: the open folder, or the app."""
|
|
if app_state.xml_root_element is None:
|
|
return None
|
|
if app_state.current_sheet_parent is not None:
|
|
return app_state.current_sheet_parent
|
|
|
|
app_element = app_state.xml_root_element.find("app")
|
|
return app_element if app_element is not None else app_state.xml_root_element
|
|
|
|
|
|
def note_kit_used(kit_name):
|
|
"""Records that a placed component needs a kit the file's <schema> does not declare.
|
|
|
|
Nothing is written to the DOM here. The <schema> block is reconciled when the file is
|
|
saved, which is the one moment that can report the complete kit list the file will
|
|
carry rather than interrupting every placement with a partial one.
|
|
|
|
Returns True when this call is what put the kit on the pending list.
|
|
"""
|
|
if app_state.xml_root_element is None:
|
|
return False
|
|
if kit_name in app_state.active_schema_kits:
|
|
return False
|
|
if kit_name in app_state.pending_schema_kits:
|
|
return False
|
|
|
|
app_state.pending_schema_kits[kit_name] = palette.resolve_checksum(kit_name)
|
|
return True
|
|
|
|
|
|
def used_kit_names():
|
|
"""Every kit the app's components need, taken from the type of each <comp>."""
|
|
names = set()
|
|
if app_state.xml_root_element is None:
|
|
return names
|
|
|
|
# A device session's DOM only holds the folders opened so far.
|
|
session = app_state.device_session
|
|
if session is not None and session.is_connected:
|
|
try:
|
|
return session.used_kit_names()
|
|
except Exception as error:
|
|
messagebox.showwarning(
|
|
"Kits",
|
|
f"Could not read every component from the device:\n{error}\n\n"
|
|
"Kits shown as used come only from the folders opened so far.",
|
|
)
|
|
|
|
for comp in app_state.xml_root_element.iter("comp"):
|
|
type_str = comp.get("type") or ""
|
|
if "::" in type_str:
|
|
names.add(type_str.split("::", 1)[0])
|
|
return names
|
|
|
|
|
|
def required_kit_names():
|
|
"""Every kit this app cannot run without: those its components name, plus their
|
|
dependencies, and the dependencies of those, all the way down.
|
|
|
|
A kit's manifest records the kits it was built against, and the device's scode needs
|
|
those just as much as the one a component names — so as far as the schema block is
|
|
concerned, being depended on by a used kit is the same as being used.
|
|
"""
|
|
unvisited = list(used_kit_names())
|
|
required = set(unvisited)
|
|
|
|
while unvisited:
|
|
kit_name = unvisited.pop()
|
|
checksum = (app_state.active_schema_kits.get(kit_name)
|
|
or app_state.pending_schema_kits.get(kit_name))
|
|
for dependency in schema.read_kit_dependencies(kit_name, checksum):
|
|
if dependency not in required:
|
|
required.add(dependency)
|
|
unvisited.append(dependency)
|
|
|
|
return required
|
|
|
|
|
|
def declare_used_kits():
|
|
"""Adds a <kit> entry for every kit the app needs that the <schema> block does not list.
|
|
|
|
The DOM is the authority on what is used, not the pending list, so a placement that
|
|
was later undone cannot leave a kit declared for nothing. Pending entries only supply
|
|
the checksum of the manifest the palette resolved when the component was placed.
|
|
|
|
A kit that a used kit depends on counts as needed too, so placing one component can
|
|
bring in more than one entry.
|
|
|
|
Entries already in the file are left as they are — a missing checksum there is a
|
|
deliberate tolerance on load, not something to backfill.
|
|
|
|
Returns the <kit> elements that were added, in the order they were written.
|
|
"""
|
|
if app_state.xml_root_element is None:
|
|
return []
|
|
|
|
schema_elem = app_state.xml_root_element.find("schema")
|
|
if schema_elem is None:
|
|
return []
|
|
|
|
declared = {kit.get("name") for kit in schema_elem.findall("kit")}
|
|
|
|
added = []
|
|
for name in sorted(required_kit_names() - declared):
|
|
checksum = app_state.pending_schema_kits.get(name) or palette.resolve_checksum(name)
|
|
kit_element = insert_kit_element(schema_elem, name, checksum)
|
|
|
|
app_state.active_schema_kits[name] = checksum
|
|
added.append(kit_element)
|
|
|
|
return added
|
|
|
|
|
|
def indent_kit_entry(schema_elem, kit_element):
|
|
"""Gives a freshly inserted <kit> the surrounding block's line break and indent.
|
|
|
|
ElementTree keeps whitespace in each element's ``tail``, so an inserted element lands
|
|
flush against its neighbour unless the tails are handed on: the new entry takes over
|
|
the tail that used to close the block, and the entry before it takes the tail that
|
|
separates two kits.
|
|
|
|
Callers that may have to undo the insertion snapshot the whole block instead of
|
|
tracking these tails, so nothing is reported back.
|
|
"""
|
|
kits = [child for child in schema_elem if child.tag == "kit" and child is not kit_element]
|
|
|
|
if not kits:
|
|
# A blank block keeps whatever whitespace it had as the new entry's closer, so an
|
|
# all-on-one-line <schema></schema> still gains a line break rather than losing one.
|
|
existing_text = schema_elem.text or ""
|
|
kit_element.tail = existing_text if existing_text.strip() == "" and existing_text else "\n"
|
|
schema_elem.text = "\n "
|
|
return
|
|
|
|
previous = kits[-1]
|
|
kit_element.tail = previous.tail
|
|
previous.tail = kits[-2].tail if len(kits) >= 2 else "\n "
|
|
|
|
|
|
def insert_kit_element(schema_elem, kit_name, checksum):
|
|
"""Builds a <kit> entry, places it after the last existing one, and returns it.
|
|
|
|
The block is not alphabetical — sys comes first by convention — so appending is safer
|
|
than guessing a sort position.
|
|
"""
|
|
kit_element = ET.Element("kit")
|
|
kit_element.set("name", kit_name)
|
|
if checksum:
|
|
kit_element.set("checksum", checksum)
|
|
|
|
insert_at = len(list(schema_elem))
|
|
for index, existing in enumerate(list(schema_elem)):
|
|
if existing.tag == "kit":
|
|
insert_at = index + 1
|
|
|
|
schema_elem.insert(insert_at, kit_element)
|
|
indent_kit_entry(schema_elem, kit_element)
|
|
return kit_element
|
|
|
|
|
|
def remove_kit_element(schema_elem, kit_element):
|
|
"""Takes a <kit> entry out of the block, leaving its line breaks tidy.
|
|
|
|
Whitespace lives in the tail of the element before it, so the entry that closes the
|
|
block is the one carrying the break that puts </schema> on its own line: when that is
|
|
the entry going away, its tail passes back to whatever now ends the block.
|
|
"""
|
|
kits = [child for child in schema_elem if child.tag == "kit"]
|
|
|
|
if kits and kits[-1] is kit_element:
|
|
if len(kits) >= 2:
|
|
kits[-2].tail = kit_element.tail
|
|
else:
|
|
schema_elem.text = kit_element.tail
|
|
|
|
schema_elem.remove(kit_element)
|
|
|
|
|
|
def format_kit_row(marker, kit_name, checksum):
|
|
"""One line of the save-time kit report: name, version and checksum.
|
|
|
|
Version comes from the .kit filenames in Sedona home rather than from the file, which
|
|
records only a checksum; '(unknown)' means this machine holds no .kit for that build.
|
|
"""
|
|
version = palette.version_for_checksum(kit_name, checksum) or "(unknown)"
|
|
return f" {marker} {kit_name} version {version} checksum {checksum or '(none)'}"
|
|
|
|
|
|
def describe_schema_kits(added_elements):
|
|
"""The save-time report: the kits just added, then every kit the file will declare."""
|
|
schema_elem = app_state.xml_root_element.find("schema")
|
|
added_names = [kit.get("name") for kit in added_elements]
|
|
|
|
kit_word = "kit is" if len(added_names) == 1 else "kits are"
|
|
lines = [
|
|
f"The following {kit_word} needed by this app — by a component, or by a kit one "
|
|
"of them depends on — but not declared in its schema, and has been added to it:",
|
|
"",
|
|
]
|
|
lines.extend(
|
|
format_kit_row("+", name, app_state.active_schema_kits.get(name))
|
|
for name in added_names
|
|
)
|
|
lines += [
|
|
"",
|
|
"The device's scode must also contain "
|
|
+ ", ".join(added_names)
|
|
+ " or the application will not run.",
|
|
"",
|
|
"Schema kits after this save:",
|
|
"",
|
|
]
|
|
|
|
for kit in schema_elem.findall("kit"):
|
|
name = kit.get("name")
|
|
if not name:
|
|
continue
|
|
marker = "+" if name in added_names else "•"
|
|
lines.append(format_kit_row(marker, name, kit.get("checksum")))
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def schema_kit_checksums():
|
|
"""The file's declared kits as an ordered list of (name, checksum), checksum may be None."""
|
|
if app_state.xml_root_element is None:
|
|
return []
|
|
|
|
schema_elem = app_state.xml_root_element.find("schema")
|
|
if schema_elem is None:
|
|
return []
|
|
|
|
return [
|
|
(kit.get("name"), kit.get("checksum"))
|
|
for kit in schema_elem.findall("kit")
|
|
if kit.get("name")
|
|
]
|
|
|
|
|
|
def unresolved_component_types():
|
|
"""Types placed in the app that the current registry cannot describe, sorted.
|
|
|
|
Repinning a kit to another build can drop a type the app still uses, which would leave
|
|
those components on the sheet with no slots. Checked after every schema change so the
|
|
change can be refused rather than silently emptying a box.
|
|
"""
|
|
if app_state.xml_root_element is None:
|
|
return []
|
|
|
|
missing = set()
|
|
for comp in app_state.xml_root_element.iter("comp"):
|
|
type_str = comp.get("type")
|
|
if type_str and type_str not in app_state.final_components_registry:
|
|
missing.add(type_str)
|
|
return sorted(missing)
|
|
|
|
|
|
def apply_kit_selections(selections, removals=(), label="update kits"):
|
|
"""Repins, adds and drops <kit> entries in one go.
|
|
|
|
``selections`` is an iterable of (kit name, checksum) the block should end up
|
|
declaring; ``removals`` is kit names it should stop declaring. Entries that already
|
|
match are skipped, so the return value describes only what really moved.
|
|
|
|
The whole set is applied, then the registry is recompiled once and checked: if a
|
|
manifest is missing, or a type the app uses no longer resolves, the block goes back to
|
|
how it was and the reason is returned instead. Nothing is half-applied.
|
|
|
|
Returns (changes, error) where changes is a list of (kit name, old checksum, new
|
|
checksum) — a new checksum of None meaning the entry was dropped — and error is None
|
|
on success.
|
|
"""
|
|
if app_state.xml_root_element is None:
|
|
return [], "No file is open."
|
|
|
|
schema_elem = app_state.xml_root_element.find("schema")
|
|
if schema_elem is None:
|
|
return [], "This file has no <schema> block to record kits in."
|
|
|
|
before = snapshot_children(schema_elem)
|
|
# Whatever the file could not describe already is not this change's doing, and holding
|
|
# it against the change would lock the kit dialog on the very file that needs it.
|
|
before_unresolved = set(unresolved_component_types())
|
|
|
|
# A kit some component needs, directly or through another kit's dependencies, cannot
|
|
# be dropped whatever the caller asks: the app would not run without it.
|
|
protected = sorted(set(removals) & required_kit_names())
|
|
if protected:
|
|
return [], (
|
|
"No kit was changed.\n\nThese kits cannot be removed, because this app "
|
|
"needs them:\n\n"
|
|
+ "\n".join(f" • {name}" for name in protected)
|
|
)
|
|
|
|
changes = []
|
|
for kit_name in removals:
|
|
for kit in schema_elem.findall("kit"):
|
|
if kit.get("name") == kit_name:
|
|
changes.append((kit_name, kit.get("checksum"), None))
|
|
remove_kit_element(schema_elem, kit)
|
|
app_state.active_schema_kits.pop(kit_name, None)
|
|
break
|
|
|
|
for kit_name, checksum in selections:
|
|
if not kit_name or not checksum:
|
|
continue
|
|
|
|
existing = None
|
|
for kit in schema_elem.findall("kit"):
|
|
if kit.get("name") == kit_name:
|
|
existing = kit
|
|
break
|
|
|
|
if existing is None:
|
|
insert_kit_element(schema_elem, kit_name, checksum)
|
|
changes.append((kit_name, None, checksum))
|
|
elif existing.get("checksum") != checksum:
|
|
old_checksum = existing.get("checksum")
|
|
existing.set("checksum", checksum)
|
|
changes.append((kit_name, old_checksum, checksum))
|
|
|
|
if not changes:
|
|
return [], None
|
|
|
|
def abandon(reason):
|
|
restore_children(schema_elem, before)
|
|
undo.reload_schema_registry()
|
|
return [], reason
|
|
|
|
# The palette resolves a kit through the checksum the schema pins, so its cached
|
|
# manifests describe the builds that were pinned a moment ago.
|
|
palette.invalidate()
|
|
try:
|
|
schema.load_schema_kits_and_manifests()
|
|
except schema.MissingManifestError as missing:
|
|
return abandon(
|
|
"No kit was changed.\n\n" + schema.describe_missing_manifests(missing)
|
|
)
|
|
|
|
unresolved = sorted(set(unresolved_component_types()) - before_unresolved)
|
|
if unresolved:
|
|
type_list = "\n".join(f" • {name}" for name in unresolved)
|
|
return abandon(
|
|
"No kit was changed.\n\nIllegal components:\n\n"
|
|
f"{type_list}\n\nFor correcting:\n\n- Delete those components first\nor\n"
|
|
"- Add related kits."
|
|
)
|
|
|
|
for kit_name, _, checksum in changes:
|
|
app_state.pending_schema_kits.pop(kit_name, None)
|
|
|
|
undo.push(undo.SchemaBlockCommand(
|
|
schema_elem, before, snapshot_children(schema_elem), label=label
|
|
))
|
|
|
|
app_state.mark_dirty()
|
|
refresh_treeview_from_dom(app_state.xml_root_element)
|
|
wiresheet.render_current_sheet()
|
|
|
|
return changes, None
|
|
|
|
|
|
def reconcile_schema_kits_before_save():
|
|
"""Brings <schema> in step with the components, warning before anything is written.
|
|
|
|
Returns False when the user backs out, in which case the added entries are taken back
|
|
out and the save is abandoned.
|
|
"""
|
|
schema_elem = app_state.xml_root_element.find("schema")
|
|
before = snapshot_children(schema_elem) if schema_elem is not None else None
|
|
|
|
added = declare_used_kits()
|
|
if not added:
|
|
return True
|
|
|
|
proceed = messagebox.askokcancel(
|
|
"Kits added to schema",
|
|
describe_schema_kits(added) + "\n\nSave with these kits?",
|
|
icon=messagebox.WARNING
|
|
)
|
|
|
|
if not proceed:
|
|
restore_children(schema_elem, before)
|
|
for kit_element in added:
|
|
app_state.active_schema_kits.pop(kit_element.get("name"), None)
|
|
return False
|
|
|
|
# Declared in the file now, so nothing about them is pending any more.
|
|
for kit_element in added:
|
|
app_state.pending_schema_kits.pop(kit_element.get("name"), None)
|
|
return True
|
|
|
|
|
|
def build_config_values(full_type, overrides=None, device_values=None):
|
|
"""Builds SOX config_values list from manifest defaults for an add command.
|
|
|
|
`overrides` maps slot name to the document-form value to send instead of the default.
|
|
`device_values` maps slot name to a value as the device sent it, for putting back a
|
|
component exactly as it was; it wins over both.
|
|
Raises ValueError when a value does not fit its slot's type.
|
|
"""
|
|
from live_hooks import _coerce_value, _wire_type, wire_value
|
|
|
|
slots = app_state.final_components_registry.get(full_type, [])
|
|
config_values = []
|
|
for slot in slots:
|
|
if "c" not in slot["flags"]:
|
|
continue
|
|
if slot["type"] == "void":
|
|
continue
|
|
tid = _wire_type(slot["type"], slot["facets"].get("asStr") is True)
|
|
if device_values and slot["name"] in device_values:
|
|
config_values.append((tid, wire_value(tid, device_values[slot["name"]])))
|
|
continue
|
|
val = (overrides or {}).get(slot["name"], slot["default"])
|
|
if val is None:
|
|
val = properties.type_zero(slot["type"])
|
|
if val is None:
|
|
continue
|
|
config_values.append((tid, _coerce_value(tid, val)))
|
|
return config_values
|
|
|
|
|
|
def grid_position_below_children(parent):
|
|
"""A spot for a component added without a drop point: under the parent's lowest child.
|
|
|
|
Used when a component is placed from the tree, where there is no point on the sheet
|
|
to drop it at. Lining it up on the left margin below everything already there keeps
|
|
it clear of the existing boxes in the common case; Tidy sorts out the rest.
|
|
"""
|
|
min_step = wiresheet.layout_margin(app_state.meta_scale) // app_state.meta_scale
|
|
lowest = None
|
|
for child in parent:
|
|
if child.tag != "comp":
|
|
continue
|
|
meta_prop = child.find("prop[@name='meta']")
|
|
y = read_meta_property(meta_prop.get("val") if meta_prop is not None else None)[1]
|
|
lowest = y if lowest is None else max(lowest, y)
|
|
|
|
if lowest is None:
|
|
return min_step, min_step
|
|
return min_step, max(min_step, min(wiresheet.META_COORD_MAX, lowest + TREE_DROP_GAP))
|
|
|
|
|
|
# Meta steps between the lowest existing box and one added under it from the tree.
|
|
TREE_DROP_GAP = 8
|
|
|
|
|
|
def add_component(kit_name, type_name, grid_x, grid_y, parent=None):
|
|
"""Places a new component of kit::type at a grid position. Returns the element.
|
|
|
|
parent is the component it goes inside; None means the sheet that is open.
|
|
"""
|
|
if parent is None:
|
|
parent = current_sheet_container()
|
|
if parent is None:
|
|
messagebox.showwarning("Add Component", "Open a SAX file first.")
|
|
return None
|
|
|
|
if note_kit_used(kit_name):
|
|
# An undeclared kit brings types the registry has never resolved, and the box
|
|
# cannot be painted until it has them.
|
|
try:
|
|
schema.load_schema_kits_and_manifests()
|
|
except schema.MissingManifestError as missing:
|
|
messagebox.showerror(
|
|
"Kit manifest missing",
|
|
f"'{kit_name}' was not added.\n\n"
|
|
+ schema.describe_missing_manifests(missing)
|
|
)
|
|
# The kit comes back off the pending list and the registry is rebuilt from
|
|
# what was resolvable a moment ago, so only this placement is abandoned.
|
|
app_state.pending_schema_kits.pop(kit_name, None)
|
|
try:
|
|
schema.load_schema_kits_and_manifests()
|
|
except schema.MissingManifestError:
|
|
reset_to_startup_state()
|
|
return None
|
|
|
|
comp_name = unique_child_name(parent, type_name)
|
|
|
|
if app_state.live_mode:
|
|
full_type = f"{kit_name}::{type_name}"
|
|
try:
|
|
config_values = build_config_values(full_type)
|
|
except (ValueError, TypeError) as e:
|
|
live_hooks.warn_failure("Add Component", f"{comp_name} was not added to the device",
|
|
f"A default value in the kit manifest does not fit its slot: {e}")
|
|
return None
|
|
new_id = live_hooks.live_add_component(
|
|
parent, kit_name, type_name, comp_name, config_values or None)
|
|
if new_id is None:
|
|
live_hooks.warn_failure("Add Component", f"{comp_name} was not added to the device")
|
|
return None
|
|
comp_id = new_id
|
|
else:
|
|
comp_id = allocate_component_id()
|
|
|
|
element = ET.Element("comp")
|
|
element.set("name", comp_name)
|
|
element.set("id", str(comp_id))
|
|
element.set("type", f"{kit_name}::{type_name}")
|
|
|
|
full_type = f"{kit_name}::{type_name}"
|
|
slots = app_state.final_components_registry.get(full_type, [])
|
|
for slot in slots:
|
|
flags = slot["flags"]
|
|
if "a" in flags or "r" in flags:
|
|
continue
|
|
if slot["type"] == "void":
|
|
continue
|
|
val = slot["default"]
|
|
if val is None:
|
|
val = properties.type_zero(slot["type"])
|
|
if val is None:
|
|
continue
|
|
prop = ET.SubElement(element, "prop")
|
|
prop.set("name", slot["name"])
|
|
prop.set("val", val)
|
|
|
|
index = len(list(parent))
|
|
parent.append(element)
|
|
|
|
if app_state.live_mode:
|
|
ds = app_state.device_session
|
|
if ds is not None:
|
|
ds.update_element_from_cache(comp_id, app_state.xml_root_element)
|
|
|
|
update_meta_property(element, None, grid_x, grid_y)
|
|
|
|
position_failed = False
|
|
if app_state.live_mode:
|
|
meta_prop = element.find("prop[@name='meta']")
|
|
if meta_prop is not None:
|
|
position_failed = not live_hooks.live_set_prop_value(
|
|
element, "meta", meta_prop.get("val", "0"))
|
|
|
|
if app_state.live_mode:
|
|
undo.push(LiveComponentCommand(
|
|
parent, index, element, present=True,
|
|
label=f"add {comp_name}"
|
|
))
|
|
else:
|
|
undo.push(undo.AddComponentCommand(
|
|
parent, index, element,
|
|
label=f"add {comp_name}"
|
|
))
|
|
|
|
app_state.mark_dirty()
|
|
refresh_treeview_from_dom(app_state.xml_root_element)
|
|
wiresheet.render_current_sheet()
|
|
|
|
if position_failed:
|
|
# The component exists on the device, so it stays; only where it sits is unsaved.
|
|
live_hooks.warn_failure(
|
|
"Add Component",
|
|
f"{comp_name} was added to the device, but its position on the sheet was not "
|
|
"written, so the device still has it at its default position")
|
|
elif app_state.live_mode:
|
|
app_state.set_status(f"{comp_name} ({kit_name}::{type_name}) added to device.")
|
|
elif kit_name in app_state.pending_schema_kits:
|
|
app_state.set_status(
|
|
f"{comp_name} ({kit_name}::{type_name}) added — '{kit_name}' is not in this "
|
|
"app's schema and will be declared when you save.",
|
|
warn=True
|
|
)
|
|
else:
|
|
app_state.set_status(f"{comp_name} ({kit_name}::{type_name}) added.")
|
|
|
|
return element
|
|
|
|
|
|
def remove_link(link_element):
|
|
"""Removes a <link> element from the <links> section and repaints."""
|
|
remove_links([link_element])
|
|
|
|
|
|
def remove_links(link_elements):
|
|
"""Removes several <link> elements as one undoable step, and repaints.
|
|
|
|
On a device each link is unlinked there first; one the device refuses stays in the
|
|
DOM, and the rest still go, with the refusals named in a single warning.
|
|
"""
|
|
if app_state.xml_root_element is None:
|
|
return
|
|
|
|
links_elem = app_state.xml_root_element.find("links")
|
|
if links_elem is None:
|
|
return
|
|
|
|
commands = []
|
|
failures = []
|
|
for link_element in link_elements:
|
|
if link_element not in list(links_elem):
|
|
continue
|
|
from_ref = link_element.get("from", "")
|
|
to_ref = link_element.get("to", "")
|
|
|
|
if app_state.live_mode:
|
|
if not live_hooks.live_delete_link(*link_ends(from_ref, to_ref)):
|
|
failures.append(f"{from_ref} → {to_ref}")
|
|
continue
|
|
|
|
index = list(links_elem).index(link_element)
|
|
links_elem.remove(link_element)
|
|
|
|
from_slot = from_ref.rsplit(".", 1)[-1] if "." in from_ref else from_ref
|
|
to_slot = to_ref.rsplit(".", 1)[-1] if "." in to_ref else to_ref
|
|
label = f"delete link {from_slot}→{to_slot}"
|
|
if app_state.live_mode:
|
|
commands.append(LiveLinkCommand(links_elem, index, link_element,
|
|
present=False, label=label))
|
|
else:
|
|
commands.append(undo.DeleteLinkCommand(links_elem, index, link_element,
|
|
label=label))
|
|
|
|
# Removed in order, each index taken after the ones before it went, so the composite
|
|
# puts them back last-first and every one lands where it was.
|
|
if len(commands) == 1:
|
|
undo.push(commands[0])
|
|
elif commands:
|
|
undo.push(undo.CompositeCommand(commands, label=f"delete {len(commands)} links"))
|
|
|
|
if failures:
|
|
noun = "link" if len(failures) == 1 else "links"
|
|
verb = "was" if len(failures) == 1 else "were"
|
|
live_hooks.warn_failure(
|
|
"Delete Link",
|
|
f"The {noun} {', '.join(failures)} {verb} not removed from the device")
|
|
|
|
if not commands:
|
|
return
|
|
|
|
app_state.selected_link_tag = None
|
|
app_state.selected_link_tags.clear()
|
|
app_state.mark_dirty()
|
|
wiresheet.render_current_sheet()
|
|
|
|
|
|
def add_link(from_path_slot, to_path_slot):
|
|
"""Adds a <link> element to the <links> section of the SAX DOM."""
|
|
if app_state.xml_root_element is None:
|
|
return
|
|
|
|
links_elem = app_state.xml_root_element.find("links")
|
|
if links_elem is None:
|
|
links_elem = ET.SubElement(app_state.xml_root_element, "links")
|
|
|
|
if app_state.live_mode:
|
|
if not live_hooks.live_add_link(*link_ends(from_path_slot, to_path_slot)):
|
|
live_hooks.warn_failure("Add Link",
|
|
f"The link {from_path_slot} → {to_path_slot} was not made on the device")
|
|
return
|
|
|
|
link_elem = ET.SubElement(links_elem, "link")
|
|
link_elem.set("from", from_path_slot)
|
|
link_elem.set("to", to_path_slot)
|
|
|
|
from_slot = from_path_slot.rsplit(".", 1)[-1] if "." in from_path_slot else from_path_slot
|
|
to_slot = to_path_slot.rsplit(".", 1)[-1] if "." in to_path_slot else to_path_slot
|
|
if app_state.live_mode:
|
|
undo.push(LiveLinkCommand(
|
|
links_elem, list(links_elem).index(link_elem), link_elem, present=True,
|
|
label=f"add link {from_slot}→{to_slot}"
|
|
))
|
|
else:
|
|
undo.push(undo.AddLinkCommand(
|
|
links_elem, link_elem,
|
|
label=f"add link {from_slot}→{to_slot}"
|
|
))
|
|
|
|
app_state.mark_dirty()
|
|
wiresheet.render_current_sheet()
|
|
|
|
|
|
# Meta steps a copy sits right of and below its original, (x, y).
|
|
DUPLICATE_OFFSET = (1, 1)
|
|
|
|
|
|
def _duplicate_tops(elements):
|
|
"""The components to copy, in document order, without any held by another of them.
|
|
|
|
A component inside one that is copied already comes along with it; copying it again
|
|
would put a second copy next to the original.
|
|
"""
|
|
root = app_state.xml_root_element
|
|
order = {element: i for i, element in enumerate(root.iter())}
|
|
chosen = []
|
|
for element in elements:
|
|
# The app cannot be copied, and a "..." stand-in is not a component at all.
|
|
if element.tag != "comp" or not element.get("type") or element not in order:
|
|
continue
|
|
if element not in chosen:
|
|
chosen.append(element)
|
|
tops = [e for e in chosen
|
|
if not any(other is not e and e in other.iter() for other in chosen)]
|
|
return sorted(tops, key=order.get)
|
|
|
|
|
|
def _meta_val(element):
|
|
meta_prop = element.find("prop[@name='meta']")
|
|
return meta_prop.get("val") if meta_prop is not None else None
|
|
|
|
|
|
def _moved_meta(meta_str, dx, dy):
|
|
"""meta moved by (dx, dy) steps, clamped to the sheet, as an unsigned int."""
|
|
x, y = read_meta_property(meta_str)[:2]
|
|
scratch = ET.Element("comp")
|
|
update_meta_property(scratch, meta_str,
|
|
min(wiresheet.META_COORD_MAX, x + dx),
|
|
min(wiresheet.META_COORD_MAX, y + dy))
|
|
return int(scratch.find("prop").get("val"))
|
|
|
|
|
|
def _ref_inside(ref, path):
|
|
return ref.startswith(path + ".") or ref.startswith(path + "/")
|
|
|
|
|
|
def duplicate_components(elements):
|
|
"""Copies components, with whatever they hold, next to the originals.
|
|
|
|
Links running between the copied components are copied too, so a group of linked
|
|
blocks comes out as a second working group. Links to anything left behind are not:
|
|
the copy would otherwise drive, or be driven by, the original's neighbours. Each
|
|
copy is named after its original with the next free number, and the whole duplicate
|
|
is one undo step. Returns the new top elements.
|
|
"""
|
|
root = app_state.xml_root_element
|
|
if root is None:
|
|
return []
|
|
tops = _duplicate_tops(elements)
|
|
if not tops:
|
|
return []
|
|
|
|
parents = [find_parent_element(root, e) for e in tops]
|
|
taken = {}
|
|
names = []
|
|
for parent, element in zip(parents, tops):
|
|
if id(parent) not in taken:
|
|
taken[id(parent)] = {c.get("name") for c in parent if c.tag == "comp"}
|
|
name = duplicate_name(element.get("name", "c"), taken[id(parent)])
|
|
taken[id(parent)].add(name)
|
|
names.append(name)
|
|
dx, dy = DUPLICATE_OFFSET
|
|
|
|
if app_state.live_mode:
|
|
copies = _duplicate_on_device(tops, parents, names, dx, dy)
|
|
else:
|
|
copies = _duplicate_in_file(tops, parents, names, dx, dy)
|
|
if not copies:
|
|
return []
|
|
|
|
app_state.mark_dirty()
|
|
refresh_treeview_from_dom(root)
|
|
wiresheet.render_current_sheet()
|
|
return copies
|
|
|
|
|
|
def _duplicate_label(tops):
|
|
if len(tops) == 1:
|
|
return f"duplicate {tops[0].get('name', '?')}"
|
|
return f"duplicate {len(tops)} components"
|
|
|
|
|
|
def _duplicate_in_file(tops, parents, names, dx, dy):
|
|
root = app_state.xml_root_element
|
|
path_map = []
|
|
commands = []
|
|
copies = []
|
|
for parent, element, name in zip(parents, tops, names):
|
|
old_path = get_component_path(element, root)
|
|
copy_elem = copy.deepcopy(element)
|
|
copy_elem.set("name", name)
|
|
x, y = read_meta_property(_meta_val(element))[:2]
|
|
update_meta_property(copy_elem, _meta_val(copy_elem),
|
|
min(wiresheet.META_COORD_MAX, x + dx),
|
|
min(wiresheet.META_COORD_MAX, y + dy))
|
|
index = len(list(parent))
|
|
parent.append(copy_elem)
|
|
# In the document first, so each new id is checked against the ones before it.
|
|
# The id a copy still carries is its original's, which stays taken either way.
|
|
for comp in copy_elem.iter("comp"):
|
|
comp.set("id", str(allocate_component_id()))
|
|
path_map.append((old_path, get_component_path(copy_elem, root)))
|
|
commands.append(undo.AddComponentCommand(parent, index, copy_elem))
|
|
copies.append(copy_elem)
|
|
|
|
def mapped(ref):
|
|
for old_path, new_path in path_map:
|
|
if _ref_inside(ref, old_path):
|
|
return new_path + ref[len(old_path):]
|
|
return None
|
|
|
|
links_elem = root.find("links")
|
|
if links_elem is not None:
|
|
for link in list(links_elem):
|
|
if link.tag != "link":
|
|
continue
|
|
new_from = mapped(link.get("from", ""))
|
|
new_to = mapped(link.get("to", ""))
|
|
if new_from is None or new_to is None:
|
|
continue
|
|
new_link = ET.SubElement(links_elem, "link")
|
|
new_link.set("from", new_from)
|
|
new_link.set("to", new_to)
|
|
commands.append(undo.AddLinkCommand(links_elem, new_link))
|
|
|
|
undo.push(undo.CompositeCommand(commands, label=_duplicate_label(tops)))
|
|
link_count = len(commands) - len(copies)
|
|
app_state.set_status(f"Duplicated as {', '.join(names)}"
|
|
+ (f", with {link_count} link(s)." if link_count else "."))
|
|
return copies
|
|
|
|
|
|
def _duplicate_on_device(tops, parents, names, dx, dy):
|
|
"""The device makes the copies from what it holds, not from the DOM.
|
|
|
|
The DOM only knows what has been opened: a folder's children may be "..." stand-ins
|
|
with no config of their own. Each original is read back in full, as a delete reads
|
|
it to be undone, and the copies are added from that.
|
|
"""
|
|
snapshots = []
|
|
for element in tops:
|
|
snapshot = live_hooks.read_subtree(element)
|
|
if snapshot is None:
|
|
live_hooks.warn_failure("Duplicate",
|
|
f"{element.get('name', '?')} could not be read from the device")
|
|
return []
|
|
snapshots.append(snapshot)
|
|
|
|
copied_ids = set()
|
|
|
|
def collect(node):
|
|
copied_ids.add(node["id"])
|
|
for child in node["children"]:
|
|
collect(child)
|
|
|
|
for snapshot in snapshots:
|
|
collect(snapshot["tree"])
|
|
|
|
links = []
|
|
for snapshot in snapshots:
|
|
for from_end, from_slot, to_end, to_slot in snapshot["links"]:
|
|
ends = (live_hooks._link_end_id(from_end), from_slot,
|
|
live_hooks._link_end_id(to_end), to_slot)
|
|
if ends[0] in copied_ids and ends[2] in copied_ids and ends not in links:
|
|
links.append(ends)
|
|
|
|
plans = []
|
|
for parent, snapshot, name in zip(parents, snapshots, names):
|
|
meta = snapshot["tree"]["values"].get("meta")
|
|
new_meta = None
|
|
if meta is not None:
|
|
new_meta = _moved_meta(str(meta), dx, dy)
|
|
if new_meta >= 1 << 31:
|
|
new_meta -= 1 << 32 # an int slot, as the device sent it: signed
|
|
plans.append((parent, snapshot["tree"], name, new_meta))
|
|
|
|
created, new_ids, complete = live_hooks.duplicate_subtrees(plans, links,
|
|
build_config_values)
|
|
if not created:
|
|
live_hooks.warn_failure("Duplicate", "Nothing was duplicated on the device")
|
|
return []
|
|
|
|
ds = app_state.device_session
|
|
client = ds.sox_client
|
|
commands = []
|
|
copies = []
|
|
for parent, new_id in created:
|
|
comp = client.cache_get(new_id)
|
|
if comp is None:
|
|
continue
|
|
copy_elem = ds._build_comp_element(parent, comp)
|
|
commands.append(LiveComponentCommand(
|
|
parent, list(parent).index(copy_elem), copy_elem, present=True))
|
|
copies.append(copy_elem)
|
|
|
|
root = app_state.xml_root_element
|
|
links_elem = root.find("links")
|
|
if links_elem is None:
|
|
links_elem = ET.SubElement(root, "links")
|
|
if complete:
|
|
for from_id, from_slot, to_id, to_slot in links:
|
|
from_comp = client.cache_get(new_ids[from_id])
|
|
to_comp = client.cache_get(new_ids[to_id])
|
|
if from_comp is None or to_comp is None:
|
|
continue
|
|
from_name = ds._slot_name_by_id(from_comp.kit_id, from_comp.type_id, from_slot)
|
|
to_name = ds._slot_name_by_id(to_comp.kit_id, to_comp.type_id, to_slot)
|
|
from_path = ds._comp_path(from_comp.id)
|
|
to_path = ds._comp_path(to_comp.id)
|
|
if from_name and to_name and from_path and to_path:
|
|
ET.SubElement(links_elem, "link",
|
|
**{"from": f"{from_path}.{from_name}",
|
|
"to": f"{to_path}.{to_name}"})
|
|
|
|
# Deleting a copy from the device takes its links with it, so the copies alone
|
|
# undo the whole duplicate.
|
|
undo.push(undo.CompositeCommand(commands, label=_duplicate_label(tops)))
|
|
|
|
if not complete:
|
|
live_hooks.warn_failure("Duplicate", "The duplicate did not fully reach the device")
|
|
else:
|
|
app_state.set_status(f"Duplicated as {', '.join(names)} on the device"
|
|
+ (f", with {len(links)} link(s)." if links else "."))
|
|
return copies
|
|
|
|
|
|
def execute_cascading_deletion(comp_element):
|
|
"""Shared application logic to wipe an element, its children, and related links from DOM and runtime lists."""
|
|
comp_name = comp_element.get("name", "Unknown")
|
|
comp_id = comp_element.get("id", "??")
|
|
|
|
descendant_comps = [c for c in comp_element.iter() if c.tag == "comp" and c != comp_element]
|
|
descendant_count = len(descendant_comps)
|
|
|
|
warning_msg = f"Are you sure you want to remove component '{comp_name}' [id: {comp_id}]?"
|
|
if descendant_count > 0:
|
|
warning_msg += f"\n\n⚠️ WARNING: This component contains {descendant_count} nested sub-component(s) under it. Deleting it will purge ALL of them recursively."
|
|
|
|
confirm = messagebox.askyesno("Confirm Cascading Delete", warning_msg)
|
|
if not confirm:
|
|
return False
|
|
|
|
snapshot = None
|
|
if app_state.live_mode:
|
|
# Read before it goes: once deleted, the device has nothing left to read back.
|
|
snapshot = live_hooks.read_subtree(comp_element)
|
|
unreadable = live_hooks.last_error if snapshot is None else ""
|
|
if not live_hooks.live_delete_component(comp_element):
|
|
live_hooks.warn_failure("Delete Component", f"{comp_name} was not deleted from the device")
|
|
return False
|
|
|
|
all_elements_to_prune = [comp_element] + descendant_comps
|
|
for elem in all_elements_to_prune:
|
|
if elem in app_state.active_components_list:
|
|
app_state.active_components_list.remove(elem)
|
|
|
|
link_records = detach_links_touching(comp_element)
|
|
|
|
xml_parent = None
|
|
comp_index = 0
|
|
if app_state.xml_root_element is not None:
|
|
xml_parent = find_parent_element(app_state.xml_root_element, comp_element)
|
|
if xml_parent is not None:
|
|
comp_index = list(xml_parent).index(comp_element)
|
|
xml_parent.remove(comp_element)
|
|
|
|
if xml_parent is not None and app_state.live_mode:
|
|
if snapshot is not None:
|
|
undo.push(LiveComponentCommand(
|
|
xml_parent, comp_index, comp_element, present=False,
|
|
snapshot=snapshot, link_records=link_records,
|
|
label=f"delete {comp_name}"
|
|
))
|
|
else:
|
|
app_state.set_status(
|
|
f"{comp_name} was deleted from the device, but this delete cannot be "
|
|
f"undone: {unreadable}", warn=True)
|
|
elif xml_parent is not None:
|
|
undo.push(undo.DeleteComponentCommand(
|
|
xml_parent, comp_index, comp_element, link_records,
|
|
label=f"delete {comp_name}"
|
|
))
|
|
|
|
target_canvas_tag = app_state.canvas_tag_by_element.get(comp_element)
|
|
if target_canvas_tag and target_canvas_tag == app_state.selected_canvas_tag:
|
|
app_state.selected_canvas_tag = None
|
|
app_state.selected_canvas_tags.discard(target_canvas_tag)
|
|
|
|
if target_canvas_tag:
|
|
app_state.canvas.delete(target_canvas_tag)
|
|
|
|
app_state.mark_dirty()
|
|
|
|
refresh_treeview_from_dom(app_state.xml_root_element)
|
|
wiresheet.render_components(app_state.active_components_list)
|
|
return True
|
|
|
|
|
|
def link_ends(from_ref, to_ref):
|
|
"""(from_elem, from_slot, to_elem, to_slot) for a link's "path.slot" ends."""
|
|
from_path, from_slot = from_ref.rsplit(".", 1) if "." in from_ref else ("", from_ref)
|
|
to_path, to_slot = to_ref.rsplit(".", 1) if "." in to_ref else ("", to_ref)
|
|
root = app_state.xml_root_element
|
|
return (find_component_by_path(root, from_path), from_slot,
|
|
find_component_by_path(root, to_path), to_slot)
|
|
|
|
|
|
def detach_links_touching(comp_element):
|
|
"""Takes every <link> into or out of `comp_element`'s subtree out of the DOM.
|
|
|
|
Returns (links_parent, index, link) records. Positions are recorded before anything
|
|
is removed, so putting them back in ascending order lands each on its old index.
|
|
"""
|
|
root = app_state.xml_root_element
|
|
if root is None:
|
|
return []
|
|
links_elem = root.find("links")
|
|
if links_elem is None:
|
|
return []
|
|
|
|
comp_path = get_component_path(comp_element, root)
|
|
link_records = []
|
|
for index, link in enumerate(list(links_elem)):
|
|
if link.tag != "link":
|
|
continue
|
|
l_from = link.get("from", "")
|
|
l_to = link.get("to", "")
|
|
from_match = l_from.startswith(comp_path + ".") or l_from.startswith(comp_path + "/")
|
|
to_match = l_to.startswith(comp_path + ".") or l_to.startswith(comp_path + "/")
|
|
if from_match or to_match:
|
|
link_records.append((links_elem, index, link))
|
|
|
|
for links_parent, _, link in link_records:
|
|
links_parent.remove(link)
|
|
return link_records
|
|
|
|
|
|
class LiveComponentCommand:
|
|
"""A component added to or deleted from the device, undone and redone through Sox.
|
|
|
|
Taking it away reads its whole subtree back first (config, children, links) and
|
|
then deletes it; putting it back adds all of that again. The device gives every
|
|
component a new id when it comes back, and the DOM elements take those ids on, so
|
|
later commands holding the same elements still reach the right components.
|
|
"""
|
|
|
|
def __init__(self, parent, index, element, present, snapshot=None, link_records=(),
|
|
label="add"):
|
|
# present: whether the component is on the device now. snapshot: from
|
|
# live_hooks.read_subtree, taken when it was last taken away.
|
|
self.parent = parent
|
|
self.index = index
|
|
self.element = element
|
|
self.present = present
|
|
self.snapshot = snapshot
|
|
self.link_records = list(link_records)
|
|
self.label = label
|
|
|
|
def undo(self):
|
|
self._flip()
|
|
|
|
def redo(self):
|
|
self._flip()
|
|
|
|
def _flip(self):
|
|
if self.present:
|
|
self._take_away()
|
|
else:
|
|
self._put_back()
|
|
|
|
def _take_away(self):
|
|
name = self.element.get("name", "?")
|
|
snapshot = live_hooks.read_subtree(self.element)
|
|
if snapshot is None:
|
|
raise undo.LiveUndoError(f"{name} was not deleted from the device")
|
|
if not live_hooks.live_delete_component(self.element):
|
|
raise undo.LiveUndoError(f"{name} was not deleted from the device")
|
|
|
|
self.snapshot = snapshot
|
|
self.present = False
|
|
self.link_records = detach_links_touching(self.element)
|
|
if self.element in list(self.parent):
|
|
self.index = list(self.parent).index(self.element)
|
|
self.parent.remove(self.element)
|
|
|
|
def _put_back(self):
|
|
name = self.element.get("name", "?")
|
|
created, complete = live_hooks.recreate_subtree(
|
|
self.parent, self.snapshot, build_config_values)
|
|
if not created:
|
|
raise undo.LiveUndoError(f"{name} was not put back on the device")
|
|
|
|
self.present = True
|
|
if self.element not in list(self.parent):
|
|
self.parent.insert(self.index, self.element)
|
|
for links_parent, index, link in sorted(self.link_records, key=lambda r: r[1]):
|
|
if link not in list(links_parent):
|
|
links_parent.insert(index, link)
|
|
|
|
ds = app_state.device_session
|
|
if ds is not None:
|
|
ds.update_element_from_cache(live_hooks._comp_id(self.element),
|
|
app_state.xml_root_element)
|
|
if not complete:
|
|
raise undo.LiveUndoError(f"{name} was not put back on the device in full")
|
|
|
|
|
|
class LiveLinkCommand:
|
|
"""A link made or removed on the device, undone and redone through Sox."""
|
|
|
|
def __init__(self, links_parent, index, link_element, present, label="add link"):
|
|
self.links_parent = links_parent
|
|
self.index = index
|
|
self.link_element = link_element
|
|
self.present = present
|
|
self.label = label
|
|
|
|
def undo(self):
|
|
self._flip()
|
|
|
|
def redo(self):
|
|
self._flip()
|
|
|
|
def _flip(self):
|
|
if self.present:
|
|
self._unlink()
|
|
else:
|
|
self._link()
|
|
|
|
def _describe(self):
|
|
return f"{self.link_element.get('from', '')} → {self.link_element.get('to', '')}"
|
|
|
|
def _ends(self):
|
|
return link_ends(self.link_element.get("from", ""), self.link_element.get("to", ""))
|
|
|
|
def _unlink(self):
|
|
if not live_hooks.live_delete_link(*self._ends()):
|
|
raise undo.LiveUndoError(f"The link {self._describe()} was not removed from the device")
|
|
self.present = False
|
|
if self.link_element in list(self.links_parent):
|
|
self.index = list(self.links_parent).index(self.link_element)
|
|
self.links_parent.remove(self.link_element)
|
|
|
|
def _link(self):
|
|
if not live_hooks.live_add_link(*self._ends()):
|
|
raise undo.LiveUndoError(f"The link {self._describe()} was not made on the device")
|
|
self.present = True
|
|
if self.link_element not in list(self.links_parent):
|
|
self.links_parent.insert(self.index, self.link_element)
|