ArdaandClaude Opus 5.5
b98fe828da
Undo and redo now work against a connected device; each step is sent there through Sox. Deleting a component first reads its whole subtree back (config, unloaded children, links, place among siblings) so undo can recreate it; recreated components take new device ids, which the DOM elements pick up so later steps still reach them. Binary Buf config (a user's cred) is read with read_prop, since the cache decodes every Buf as a Str. Moves, tidy layout, and Properties / page editor / user edits record only what the device accepted. A step that does not fully reach the device warns and clears the history. History is also cleared on every view change in live mode, as opening a sheet rebuilds the DOM from the device. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
505 lines
18 KiB
Python
505 lines
18 KiB
Python
"""
|
|
Write-through interception layer for live device mode.
|
|
|
|
Checks app_state.live_mode and routes mutations through DeviceSession/SoxClient.
|
|
In offline mode, falls through to the original DOM-only code paths.
|
|
"""
|
|
|
|
import base64
|
|
import binascii
|
|
from tkinter import messagebox
|
|
|
|
import app_state
|
|
from sox import BUF_ID, CONFIG, STR_ID, type_id_for_slot
|
|
|
|
|
|
# Why the last live_* call failed, in words the user can act on. Every failing call sets
|
|
# it, so a caller can put it in front of the user with warn_failure().
|
|
last_error = ""
|
|
|
|
|
|
def is_live():
|
|
return (app_state.live_mode
|
|
and app_state.device_session is not None
|
|
and app_state.device_session.is_connected)
|
|
|
|
|
|
def _fail(reason):
|
|
global last_error
|
|
last_error = reason
|
|
return None
|
|
|
|
|
|
def _session():
|
|
"""The connected DeviceSession, or None with last_error saying why not."""
|
|
if not is_live():
|
|
_fail("The device is not connected.")
|
|
return None
|
|
return app_state.device_session
|
|
|
|
|
|
def warn_failure(title, what, detail=None):
|
|
"""Tells the user, in a dialog and the status bar, that `what` did not reach the device.
|
|
|
|
`detail` defaults to last_error. Nothing retries a write that failed, so the dialog
|
|
is the only record the user gets that the editor and the device may now disagree.
|
|
"""
|
|
reason = detail if detail is not None else last_error
|
|
app_state.set_status(f"{what}: {reason}", warn=True)
|
|
messagebox.showwarning(title, f"{what}.\n\n{reason}", parent=app_state.window)
|
|
|
|
|
|
# ── Property Write ──────────────────────────────────────────────
|
|
|
|
def live_set_prop_value(element, slot_name, value_str):
|
|
"""In live mode, sends a Sox write to the device. Returns True on success."""
|
|
ds = _session()
|
|
if ds is None:
|
|
return False
|
|
comp_id = _comp_id(element)
|
|
if comp_id is None:
|
|
_fail(f"'{element.get('name', '?')}' has no component id.")
|
|
return False
|
|
|
|
slot_info = _find_slot(ds, comp_id, slot_name)
|
|
if slot_info is None:
|
|
return False
|
|
|
|
type_id = slot_info["type_id"]
|
|
try:
|
|
coerced = _coerce_value(type_id, value_str)
|
|
except (TypeError, ValueError, binascii.Error) as e:
|
|
_fail(f"'{value_str}' is not a valid value for {slot_name}: {e}")
|
|
return False
|
|
|
|
try:
|
|
ds.sox_client.write_prop(comp_id, slot_info["id"], type_id, coerced)
|
|
return True
|
|
except Exception as e:
|
|
_fail(f"The device rejected the write to {slot_name}: {e}")
|
|
return False
|
|
|
|
|
|
# ── Add Component ───────────────────────────────────────────────
|
|
|
|
def live_add_component(parent_element, kit_name, type_name, comp_name, config_values=None):
|
|
"""In live mode, sends a Sox add to the device. Returns new comp_id or None."""
|
|
ds = _session()
|
|
if ds is None:
|
|
return None
|
|
parent_id = _comp_id(parent_element)
|
|
if parent_id is None:
|
|
return _fail(f"The parent '{parent_element.get('name', '?')}' has no component id.")
|
|
|
|
kit_id = _kit_id(ds, kit_name)
|
|
if kit_id is None:
|
|
return _fail(f"Kit '{kit_name}' is not installed on the device.")
|
|
type_id_val = _type_id(ds, kit_id, type_name)
|
|
if type_id_val is None:
|
|
return _fail(f"The device's '{kit_name}' kit has no type '{type_name}'.")
|
|
|
|
try:
|
|
new_comp = ds.sox_client.add(parent_id, kit_id, type_id_val, comp_name,
|
|
list(config_values or []))
|
|
ds.after_add(new_comp.id)
|
|
return new_comp.id
|
|
except Exception as e:
|
|
return _fail(f"The device rejected the add: {e}")
|
|
|
|
|
|
# ── Delete Component ────────────────────────────────────────────
|
|
|
|
def live_delete_component(element):
|
|
"""In live mode, sends a Sox delete to the device. Returns True on success."""
|
|
ds = _session()
|
|
if ds is None:
|
|
return False
|
|
comp_id = _comp_id(element)
|
|
if comp_id is None:
|
|
_fail(f"'{element.get('name', '?')}' has no component id.")
|
|
return False
|
|
|
|
descendant_ids = []
|
|
for desc in element.iter():
|
|
if desc.tag == "comp" and desc is not element:
|
|
did = _comp_id(desc)
|
|
if did is not None:
|
|
descendant_ids.append(did)
|
|
|
|
try:
|
|
ds.sox_client.delete(comp_id)
|
|
ds.after_delete(comp_id, descendant_ids)
|
|
return True
|
|
except Exception as e:
|
|
_fail(f"The device rejected the delete: {e}")
|
|
return False
|
|
|
|
|
|
# ── Rename Component ────────────────────────────────────────────
|
|
|
|
def live_rename_component(element, new_name):
|
|
"""In live mode, sends a Sox rename to the device. Returns True on success."""
|
|
ds = _session()
|
|
if ds is None:
|
|
return False
|
|
comp_id = _comp_id(element)
|
|
if comp_id is None:
|
|
_fail(f"'{element.get('name', '?')}' has no component id.")
|
|
return False
|
|
|
|
try:
|
|
ds.sox_client.rename(comp_id, new_name)
|
|
return True
|
|
except Exception as e:
|
|
_fail(f"The device rejected the rename: {e}")
|
|
return False
|
|
|
|
|
|
# ── Link / Unlink ───────────────────────────────────────────────
|
|
|
|
def _link_ends(ds, from_element, from_slot_name, to_element, to_slot_name):
|
|
"""(from_id, from_slot_id, to_id, to_slot_id), or None with last_error set."""
|
|
if from_element is None or to_element is None:
|
|
return _fail("One end of the link is not a component in the loaded tree.")
|
|
from_id = _comp_id(from_element)
|
|
to_id = _comp_id(to_element)
|
|
if from_id is None or to_id is None:
|
|
return _fail("One end of the link has no component id.")
|
|
from_slot = _find_slot(ds, from_id, from_slot_name)
|
|
if from_slot is None:
|
|
return None
|
|
to_slot = _find_slot(ds, to_id, to_slot_name)
|
|
if to_slot is None:
|
|
return None
|
|
return from_id, from_slot["id"], to_id, to_slot["id"]
|
|
|
|
|
|
def live_add_link(from_element, from_slot_name, to_element, to_slot_name):
|
|
"""In live mode, sends a Sox link command. Returns True on success."""
|
|
ds = _session()
|
|
if ds is None:
|
|
return False
|
|
ends = _link_ends(ds, from_element, from_slot_name, to_element, to_slot_name)
|
|
if ends is None:
|
|
return False
|
|
try:
|
|
ds.sox_client.link(*ends)
|
|
return True
|
|
except Exception as e:
|
|
_fail(f"The device rejected the link: {e}")
|
|
return False
|
|
|
|
|
|
def live_delete_link(from_element, from_slot_name, to_element, to_slot_name):
|
|
"""In live mode, sends a Sox unlink command. Returns True on success."""
|
|
ds = _session()
|
|
if ds is None:
|
|
return False
|
|
ends = _link_ends(ds, from_element, from_slot_name, to_element, to_slot_name)
|
|
if ends is None:
|
|
return False
|
|
try:
|
|
ds.sox_client.unlink(*ends)
|
|
return True
|
|
except Exception as e:
|
|
_fail(f"The device rejected the unlink: {e}")
|
|
return False
|
|
|
|
|
|
# ── Invoke Action ───────────────────────────────────────────────
|
|
|
|
def live_invoke(element, slot_name, arg=None):
|
|
"""In live mode, sends a Sox invoke. Returns True on success."""
|
|
ds = _session()
|
|
if ds is None:
|
|
return False
|
|
comp_id = _comp_id(element)
|
|
if comp_id is None:
|
|
_fail(f"'{element.get('name', '?')}' has no component id.")
|
|
return False
|
|
|
|
slot_info = _find_slot(ds, comp_id, slot_name)
|
|
if slot_info is None:
|
|
return False
|
|
|
|
try:
|
|
ds.sox_client.invoke(comp_id, slot_info["id"])
|
|
return True
|
|
except Exception as e:
|
|
_fail(f"The device rejected the invoke of {slot_name}: {e}")
|
|
return False
|
|
|
|
|
|
# ── Undo: read a subtree back, put it back ──────────────────────
|
|
|
|
def read_subtree(element):
|
|
"""Everything it takes to put `element` and all it holds back on the device.
|
|
|
|
Read from the device, not the DOM: the DOM only holds what has been opened, and a
|
|
placeholder child carries neither config nor children of its own. Requests go one
|
|
component at a time, at the pace the device sets (see used_kit_names).
|
|
|
|
Returns {"tree", "links", "siblings"}, or None with last_error set. A node is
|
|
{"id", "kit_id", "type_id", "type_name", "name", "values", "element", "children"},
|
|
where "element" is the DOM element showing it, if any, so a new id can reach it.
|
|
A link end, and each of "siblings" (the parent's children, in order), is ("node",
|
|
node), ("elem", element) or ("id", comp_id): a component outside the subtree is
|
|
followed by its element when it has one, because its id changes if it is itself
|
|
deleted and put back.
|
|
"""
|
|
ds = _session()
|
|
if ds is None:
|
|
return None
|
|
comp_id = _comp_id(element)
|
|
if comp_id is None:
|
|
return _fail(f"'{element.get('name', '?')}' has no component id.")
|
|
|
|
client = ds.sox_client
|
|
root = app_state.xml_root_element
|
|
elements_by_id = {}
|
|
if root is not None:
|
|
for elem in root.iter("comp"):
|
|
if elem.get("id") is not None:
|
|
elements_by_id.setdefault(elem.get("id"), elem)
|
|
|
|
nodes_by_id = {}
|
|
|
|
def read(comp):
|
|
if comp.type_name not in app_state.final_components_registry:
|
|
raise _SnapshotError(f"No kit manifest describes {comp.type_name}, so "
|
|
f"'{comp.name}' could not be put back as it was.")
|
|
children = ds.load_children(comp.id)
|
|
if len(children) != len(comp.children):
|
|
raise _SnapshotError(f"Not every child of '{comp.name}' could be read.")
|
|
client.update([comp], CONFIG)
|
|
values = dict(comp.values)
|
|
# The cache reads every Buf as a Str, which loses bytes of a binary one (a
|
|
# user's cred); those are read again as what they are.
|
|
for slot in app_state.final_components_registry[comp.type_name]:
|
|
if ("c" in slot["flags"] and slot["type"] != "void"
|
|
and _wire_type(slot["type"], slot["facets"].get("asStr") is True) == BUF_ID):
|
|
values[slot["name"]] = client.read_prop(comp.id, slot["slot_id"], BUF_ID)
|
|
node = {
|
|
"id": comp.id,
|
|
"kit_id": comp.kit_id,
|
|
"type_id": comp.type_id,
|
|
"type_name": comp.type_name,
|
|
"name": comp.name,
|
|
"values": values,
|
|
"element": elements_by_id.get(str(comp.id)),
|
|
"children": [],
|
|
}
|
|
nodes_by_id[comp.id] = node
|
|
node["children"] = [read(child) for child in children]
|
|
return node
|
|
|
|
try:
|
|
top = client.load(comp_id)
|
|
if top is None:
|
|
return _fail(f"'{element.get('name', '?')}' could not be read from the device.")
|
|
tree = read(top)
|
|
|
|
links = []
|
|
seen = set()
|
|
for cid in list(nodes_by_id):
|
|
for lnk in client.read_links(cid):
|
|
key = (lnk.from_comp_id, lnk.from_slot_id, lnk.to_comp_id, lnk.to_slot_id)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
links.append((_link_end(lnk.from_comp_id, nodes_by_id, elements_by_id),
|
|
lnk.from_slot_id,
|
|
_link_end(lnk.to_comp_id, nodes_by_id, elements_by_id),
|
|
lnk.to_slot_id))
|
|
except _SnapshotError as e:
|
|
return _fail(str(e))
|
|
except Exception as e:
|
|
return _fail(f"'{element.get('name', '?')}' could not be read from the device: {e}")
|
|
|
|
parent = client.cache_get(top.parent)
|
|
siblings = [_link_end(cid, nodes_by_id, elements_by_id)
|
|
for cid in (parent.children if parent is not None else [])]
|
|
return {"tree": tree, "links": links, "siblings": siblings}
|
|
|
|
|
|
class _SnapshotError(Exception):
|
|
pass
|
|
|
|
|
|
def _link_end(comp_id, nodes_by_id, elements_by_id):
|
|
if comp_id in nodes_by_id:
|
|
return ("node", nodes_by_id[comp_id])
|
|
elem = elements_by_id.get(str(comp_id))
|
|
if elem is not None:
|
|
return ("elem", elem)
|
|
return ("id", comp_id)
|
|
|
|
|
|
def _link_end_id(end):
|
|
kind, ref = end
|
|
if kind == "node":
|
|
return ref["id"]
|
|
if kind == "elem":
|
|
return _comp_id(ref)
|
|
return ref
|
|
|
|
|
|
def recreate_subtree(parent_element, snapshot, build_config):
|
|
"""Adds a subtree from read_subtree back under `parent_element`, with its links.
|
|
|
|
The device hands out new ids. Each node, and the DOM element showing it, takes its
|
|
new id, so whatever refers to them later still finds them. The component goes back
|
|
to its old place among its siblings.
|
|
|
|
`build_config(type_name, device_values=...)` makes an add command's config values.
|
|
Returns (top_created, complete); when not complete, last_error says what is missing.
|
|
"""
|
|
ds = _session()
|
|
if ds is None:
|
|
return False, False
|
|
parent_id = _comp_id(parent_element)
|
|
if parent_id is None:
|
|
_fail(f"The parent '{parent_element.get('name', '?')}' has no component id.")
|
|
return False, False
|
|
|
|
client = ds.sox_client
|
|
tree = snapshot["tree"]
|
|
|
|
def create(node, under_id):
|
|
config = build_config(node["type_name"], device_values=node["values"])
|
|
comp = client.add(under_id, node["kit_id"], node["type_id"], node["name"], config)
|
|
node["id"] = comp.id
|
|
if node["element"] is not None:
|
|
node["element"].set("id", str(comp.id))
|
|
|
|
def create_children(node):
|
|
for child in node["children"]:
|
|
create(child, node["id"])
|
|
create_children(child)
|
|
|
|
try:
|
|
create(tree, parent_id)
|
|
except Exception as e:
|
|
_fail(f"The device rejected the add of '{tree['name']}': {e}")
|
|
return False, False
|
|
|
|
try:
|
|
create_children(tree)
|
|
for from_end, from_slot, to_end, to_slot in snapshot["links"]:
|
|
client.link(_link_end_id(from_end), from_slot, _link_end_id(to_end), to_slot)
|
|
|
|
# The add put it last; the device's order is the order execution runs in.
|
|
parent = client.cache_get(parent_id)
|
|
wanted = [_link_end_id(end) for end in snapshot["siblings"]]
|
|
if (parent is not None and set(wanted) == set(parent.children)
|
|
and wanted != parent.children):
|
|
client.reorder(parent_id, wanted)
|
|
complete = True
|
|
except Exception as e:
|
|
_fail(f"'{tree['name']}' is back on the device, but not all of it: {e}")
|
|
complete = False
|
|
|
|
ds.after_add(tree["id"])
|
|
return True, complete
|
|
|
|
|
|
def wire_value(type_id, raw):
|
|
"""A value as the device sent it, in the form encode_value wants for `type_id`."""
|
|
if type_id == BUF_ID:
|
|
return bytes(raw) if isinstance(raw, (bytes, bytearray)) else str(raw).encode("utf-8")
|
|
if type_id == STR_ID:
|
|
if isinstance(raw, (bytes, bytearray)):
|
|
return bytes(raw).decode("utf-8", errors="replace").rstrip("\0")
|
|
return str(raw)
|
|
return _coerce_value(type_id, str(raw))
|
|
|
|
|
|
# ── Helpers ─────────────────────────────────────────────────────
|
|
|
|
def _comp_id(element):
|
|
if element.tag == "app":
|
|
return 0
|
|
cid = element.get("id")
|
|
if cid is None:
|
|
return None
|
|
try:
|
|
return int(cid)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _find_slot(ds, comp_id, slot_name):
|
|
"""{name, id, type_id} for a slot of a cached component, or None with last_error set.
|
|
|
|
A Buf goes over the wire as a Buf only when it is binary; an asStr Buf travels as a
|
|
Str, which is how the rest of the editor already reads and writes it.
|
|
"""
|
|
comp = ds.sox_client.cache_get(comp_id)
|
|
if comp is None:
|
|
return _fail(f"Component {comp_id} is not loaded from the device.")
|
|
type_name = ds.sox_client.type_name_map.get((comp.kit_id, comp.type_id))
|
|
if type_name and type_name in app_state.final_components_registry:
|
|
for s in app_state.final_components_registry[type_name]:
|
|
if s["name"] == slot_name:
|
|
slot_type = s.get("type") or "int"
|
|
as_str = (s.get("facets") or {}).get("asStr") is True
|
|
return {"name": s["name"], "id": s["slot_id"],
|
|
"type_id": _wire_type(slot_type, as_str)}
|
|
return _fail(f"{type_name} has no slot '{slot_name}'.")
|
|
slots = ds.sox_client.slot_registry.get((comp.kit_id, comp.type_id))
|
|
if slots is None:
|
|
return _fail(f"No kit manifest describes {type_name or 'this component'}'s slots, "
|
|
"so the slot cannot be addressed on the device.")
|
|
for s in slots:
|
|
if s["name"] == slot_name:
|
|
return {"name": s["name"], "id": s["id"],
|
|
"type_id": _wire_type(s.get("type_name") or "int", s.get("is_as_str"))}
|
|
return _fail(f"{type_name or 'This component'} has no slot '{slot_name}'.")
|
|
|
|
|
|
def _wire_type(slot_type, as_str):
|
|
if slot_type in ("sys::Buf", "Buf", "buf") and not as_str:
|
|
return BUF_ID
|
|
return type_id_for_slot(slot_type)
|
|
|
|
|
|
def _kit_id(ds, kit_name):
|
|
if ds.sox_client.schema_parts:
|
|
for i, part in enumerate(ds.sox_client.schema_parts):
|
|
if part.name == kit_name:
|
|
return i
|
|
return None
|
|
|
|
|
|
def _type_id(ds, kit_id, type_name):
|
|
if kit_id is None:
|
|
return None
|
|
for (kid, tid), full_name in ds.sox_client.type_name_map.items():
|
|
if kid == kit_id and full_name.endswith(f"::{type_name}"):
|
|
return tid
|
|
return None
|
|
|
|
|
|
def _coerce_value(type_id, value_str):
|
|
"""The Python value encode_value wants for `value_str` in document form.
|
|
|
|
Raises ValueError (or binascii.Error) when the text does not fit the type.
|
|
"""
|
|
from sox import BOOL_ID, BYTE_ID, SHORT_ID, INT_ID, LONG_ID, FLOAT_ID, DOUBLE_ID, STR_ID
|
|
if type_id == BOOL_ID:
|
|
return value_str.lower() in ("true", "1", "yes")
|
|
elif type_id in (BYTE_ID, SHORT_ID, INT_ID, LONG_ID):
|
|
try:
|
|
return int(value_str)
|
|
except ValueError:
|
|
return int(value_str, 0) # meta and flags can be written as 0x...
|
|
elif type_id in (FLOAT_ID, DOUBLE_ID):
|
|
return float(value_str)
|
|
elif type_id == STR_ID:
|
|
return str(value_str)
|
|
elif type_id == BUF_ID:
|
|
if isinstance(value_str, (bytes, bytearray)):
|
|
return bytes(value_str)
|
|
return base64.b64decode(value_str or "", validate=True) # the .sax form
|
|
return value_str
|