Files
sedona_editor/sax_file.py
T
ArdaandClaude Opus 5.5 b3d96d96b6 Tidy Layout as signal flow; fixed-size wiresheet; signed Sox int writes
- Tidy Layout lays linked boxes out left to right by signal flow, in as few
  columns as the longest chain allows; feedback loops are cut where they run
  backwards, and sources sit just before the boxes they feed.
- Boxes with no link on the open sheet go to the bottom of the first column,
  alphabetically.
- A box that wires skipping its column would cross is pushed down below them.
- The wiresheet is fixed at meta's 255-step reach; outside it is light grey.
  Dragging keeps a box's top-left corner on the sheet.
- Sox write_i4 wraps to signed, so meta with x >= 128 reaches the device.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-18 10:15:22 +03:00

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.fit_sheet_region()
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.fit_sheet_region()
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)