Arda/sedona_editor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make every live edit reach the device, and warn when it doesn't

Browse Source 
In live mode the device is now written first and the document only takes
what the device accepted. Every live_hooks failure records a reason, and
warn_failure() shows it in a dialog and the status bar instead of failing
silently.

- Move, properties, OX page editor: rejected values are rolled back and listed
- Add/remove link: no longer edits the DOM alone when an end can't be resolved
- Add/delete component: report why the device refused
- Tidy Layout: sends each new position to the device, reverts refusals,
  marks the document dirty, and pushes no undo step in live mode
- User Manager: add (one Sox add carrying cred/perm/prov), edit and delete
  now go to the device
- Binary Buf slots (e.g. User.cred) are sent as Buf bytes, not as Str text
- Bump version to 0.0.0.136

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Arda
2026-10-02 20:16:00 +03:00
co-authored by Claude Opus 5.5
parent f15fc42363
commit a2d4be6ea8
8 changed files with 312 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
live_hooks.py
+129 -71
View File
@@ -5,8 +5,17 @@ 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 type_id_for_slot
from sox import BUF_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():
@@ -15,30 +24,59 @@ def is_live():
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. Live mode has no undo, 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."""
if not is_live():
ds = _session()
if ds is None:
return False
ds = app_state.device_session
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 = type_id_for_slot(slot_info["type_name"])
coerced = _coerce_value(type_id, value_str)
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:
app_state.set_status(f"Write failed: {e}", warn=True)
_fail(f"The device rejected the write to {slot_name}: {e}")
return False
@@ -46,43 +84,39 @@ def live_set_prop_value(element, slot_name, value_str):
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."""
if not is_live():
ds = _session()
if ds is None:
return None
ds = app_state.device_session
parent_id = _comp_id(parent_element)
if parent_id is None:
return 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 kit_id is None or type_id_val is None:
return None
sox_config = []
if config_values:
for tid, val in config_values:
sox_config.append((tid, val))
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, sox_config)
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:
app_state.set_status(f"Add failed: {e}", warn=True)
return None
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."""
if not is_live():
ds = _session()
if ds is None:
return False
ds = app_state.device_session
comp_id = _comp_id(element)
if comp_id is None:
_fail(f"'{element.get('name', '?')}' has no component id.")
return False
descendant_ids = []
@@ -97,7 +131,7 @@ def live_delete_component(element):
ds.after_delete(comp_id, descendant_ids)
return True
except Exception as e:
app_state.set_status(f"Delete failed: {e}", warn=True)
_fail(f"The device rejected the delete: {e}")
return False
@@ -105,69 +139,70 @@ def live_delete_component(element):
def live_rename_component(element, new_name):
"""In live mode, sends a Sox rename to the device. Returns True on success."""
if not is_live():
ds = _session()
if ds is None:
return False
ds = app_state.device_session
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:
app_state.set_status(f"Rename failed: {e}", warn=True)
_fail(f"The device rejected the rename: {e}")
return False
# ── Link / Unlink ───────────────────────────────────────────────
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."""
if not is_live():
return False
ds = app_state.device_session
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 False
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 from_slot is None or to_slot is None:
return False
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(from_id, from_slot["id"], to_id, to_slot["id"])
ds.sox_client.link(*ends)
return True
except Exception as e:
app_state.set_status(f"Link failed: {e}", warn=True)
_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."""
if not is_live():
ds = _session()
if ds is None:
return False
ds = app_state.device_session
from_id = _comp_id(from_element)
to_id = _comp_id(to_element)
if from_id is None or to_id is None:
ends = _link_ends(ds, from_element, from_slot_name, to_element, to_slot_name)
if ends is None:
return False
from_slot = _find_slot(ds, from_id, from_slot_name)
to_slot = _find_slot(ds, to_id, to_slot_name)
if from_slot is None or to_slot is None:
return False
try:
ds.sox_client.unlink(from_id, from_slot["id"], to_id, to_slot["id"])
ds.sox_client.unlink(*ends)
return True
except Exception as e:
app_state.set_status(f"Unlink failed: {e}", warn=True)
_fail(f"The device rejected the unlink: {e}")
return False
@@ -175,12 +210,12 @@ def live_delete_link(from_element, from_slot_name, to_element, to_slot_name):
def live_invoke(element, slot_name, arg=None):
"""In live mode, sends a Sox invoke. Returns True on success."""
if not is_live():
ds = _session()
if ds is None:
return False
ds = app_state.device_session
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)
@@ -191,7 +226,7 @@ def live_invoke(element, slot_name, arg=None):
ds.sox_client.invoke(comp_id, slot_info["id"])
return True
except Exception as e:
app_state.set_status(f"Invoke failed: {e}", warn=True)
_fail(f"The device rejected the invoke of {slot_name}: {e}")
return False
@@ -210,22 +245,38 @@ def _comp_id(element):
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 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_name": s.get("type", "int")}
"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 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 s
return None
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):
@@ -246,17 +297,24 @@ def _type_id(ds, kit_id, type_name):
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):
return int(value_str)
elif type_id == LONG_ID:
return int(value_str)
elif type_id == FLOAT_ID:
return float(value_str)
elif type_id == DOUBLE_ID:
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