ArdaandClaude Opus 4.6
2530bbd92c
Fix add component from palette in live mode: resolve <app> element as comp_id 0, fix type_id lookup to use type_name_map, set type_name on newly added SoxComponent, sync device values into DOM after add, write meta position back to device, and subscribe new component to COV. Fix delete component in live mode: clean up subscriptions for deleted component and all its descendants. Add tree root context menu with device app actions (save, hibernate, quit, restart, reboot) in live mode. Device save clears dirty state. Clean up connection lost handling to do a full disconnect. Fix slot registry to copy base slots and renumber IDs correctly. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
459 lines
18 KiB
Python
459 lines
18 KiB
Python
"""
|
|
DeviceSession — bridges SoxClient to the editor's ET.Element-based UI.
|
|
|
|
Handles lazy loading, DOM synthesis from cached SoxComponents,
|
|
COV event routing, and subscription lifecycle.
|
|
"""
|
|
|
|
import queue
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import app_state
|
|
import schema
|
|
from sox import SoxClient, SoxComponent, TREE, CONFIG, RUNTIME, LINKS, ALL
|
|
|
|
|
|
def _fmt_float(val, precision=2):
|
|
"""Format a float for display with *precision* decimal places (default 2)."""
|
|
return f"{val:.{precision}f}"
|
|
|
|
|
|
def _slot_precision(type_name, slot_name):
|
|
"""Looks up the precision facet for a slot, returning 2 as the default."""
|
|
reg = app_state.final_components_registry.get(type_name)
|
|
if reg:
|
|
for s in reg:
|
|
if s["name"] == slot_name:
|
|
p = s.get("facets", {}).get("precision")
|
|
if isinstance(p, int) and 0 <= p <= 12:
|
|
return p
|
|
break
|
|
return 2
|
|
|
|
|
|
class DeviceSession:
|
|
"""High-level session manager for a live Sedona device connection."""
|
|
|
|
def __init__(self):
|
|
self.sox_client = None
|
|
self._event_queue = queue.Queue()
|
|
self._current_subs = set() # comp_ids we subscribed for the open wiresheet
|
|
|
|
# ── Connect / Disconnect ────────────────────────────────────
|
|
|
|
def connect(self, host, port, user, password):
|
|
"""DASP connect → read schema → load root. Called from background thread."""
|
|
self.sox_client = SoxClient(host, int(port), user, password)
|
|
self.sox_client.on_event = self._on_sox_event
|
|
self.sox_client.connect()
|
|
|
|
parts = self.sox_client.read_schema()
|
|
vi = self.sox_client.read_version()
|
|
|
|
self._load_slot_registry(parts)
|
|
|
|
root_comp = self.sox_client.load(0)
|
|
# Pre-load immediate children so the tree shows the first level
|
|
if root_comp and root_comp.children:
|
|
self.load_children(root_comp.id)
|
|
return root_comp
|
|
|
|
def disconnect(self):
|
|
if self.sox_client is None:
|
|
return
|
|
try:
|
|
if self._current_subs:
|
|
comps = [self.sox_client.cache_get(cid)
|
|
for cid in self._current_subs]
|
|
comps = [c for c in comps if c is not None]
|
|
if comps:
|
|
self.sox_client.unsubscribe(comps, ALL)
|
|
except Exception:
|
|
pass
|
|
self._current_subs.clear()
|
|
try:
|
|
self.sox_client.close()
|
|
except Exception:
|
|
pass
|
|
self.sox_client = None
|
|
|
|
@property
|
|
def is_connected(self):
|
|
return self.sox_client is not None and not self.sox_client.is_closed
|
|
|
|
# ── Schema → Slot Registry ──────────────────────────────────
|
|
|
|
def _load_slot_registry(self, parts):
|
|
"""Builds slot_registry from kit manifests for the device's schema."""
|
|
registry = {}
|
|
base_slots = []
|
|
for kit_id, part in enumerate(parts):
|
|
kit_name = part.name
|
|
checksum = part.checksum
|
|
manifest = schema.find_manifest(kit_name, checksum)
|
|
if manifest is None:
|
|
continue
|
|
if kit_name == "sys":
|
|
base_slots = self._extract_component_base_slots(manifest)
|
|
self._register_kit_types(registry, kit_id, kit_name, manifest)
|
|
# Prepend inherited Component base slots (meta, etc.) to every type,
|
|
# then renumber IDs to match the wire slot index (positional in the
|
|
# flattened inherited + own list — see sedona Type.java resolveSlots).
|
|
if base_slots:
|
|
for key in registry:
|
|
existing_names = {s["name"] for s in registry[key]}
|
|
for bs in base_slots:
|
|
if bs["name"] not in existing_names:
|
|
registry[key].insert(0, dict(bs))
|
|
for i, slot in enumerate(registry[key]):
|
|
slot["id"] = i
|
|
self.sox_client.slot_registry = registry
|
|
|
|
def _extract_component_base_slots(self, manifest):
|
|
"""Extracts base slots from sys::Component that all types inherit."""
|
|
for type_elem in manifest.iter("type"):
|
|
if type_elem.get("name") == "Component":
|
|
slots = []
|
|
for slot_elem in type_elem.findall("slot"):
|
|
slot_name = slot_elem.get("name", "")
|
|
slot_id = int(slot_elem.get("id", "-1"))
|
|
slot_type = slot_elem.get("type", "int")
|
|
flags_str = slot_elem.get("flags", "")
|
|
slots.append({
|
|
"name": slot_name,
|
|
"id": slot_id,
|
|
"type_name": slot_type,
|
|
"is_config": "c" in flags_str or "p" in flags_str,
|
|
"is_runtime": "r" in flags_str,
|
|
"is_action": "a" in flags_str,
|
|
"is_operator": "o" in flags_str,
|
|
"is_as_str": slot_type == "Str",
|
|
"is_prop": "a" not in flags_str,
|
|
})
|
|
return slots
|
|
return []
|
|
|
|
def _register_kit_types(self, registry, kit_id, kit_name, manifest):
|
|
root = manifest
|
|
for type_elem in root.iter("type"):
|
|
type_name = type_elem.get("name", "")
|
|
type_id_val = int(type_elem.get("id", "-1"))
|
|
if type_id_val < 0:
|
|
continue
|
|
slots = []
|
|
for slot_elem in type_elem.findall("slot"):
|
|
slot_name = slot_elem.get("name", "")
|
|
slot_id = int(slot_elem.get("id", "-1"))
|
|
slot_type = slot_elem.get("type", "int")
|
|
flags_str = slot_elem.get("flags", "")
|
|
is_config = "c" in flags_str or "p" in flags_str
|
|
is_runtime = "r" in flags_str
|
|
is_action = "a" in flags_str
|
|
is_operator = "o" in flags_str
|
|
is_as_str = slot_type == "Str" or slot_elem.get("asStr") == "true"
|
|
slots.append({
|
|
"name": slot_name,
|
|
"id": slot_id,
|
|
"type_name": slot_type,
|
|
"is_config": is_config,
|
|
"is_runtime": is_runtime,
|
|
"is_action": is_action,
|
|
"is_operator": is_operator,
|
|
"is_as_str": is_as_str,
|
|
"is_prop": not is_action,
|
|
})
|
|
registry[(kit_id, type_id_val)] = slots
|
|
full_name = f"{kit_name}::{type_name}"
|
|
self.sox_client.type_name_map[(kit_id, type_id_val)] = full_name
|
|
# update type_name on any cached component of this type
|
|
for c in self.sox_client.cache:
|
|
if c is not None and c.kit_id == kit_id and c.type_id == type_id_val:
|
|
c.type_name = full_name
|
|
|
|
# ── Lazy Loading ────────────────────────────────────────────
|
|
|
|
def load_children(self, comp_id):
|
|
"""Loads direct children of a component (cache-aware). Returns list of SoxComponent."""
|
|
parent = self.sox_client.cache_get(comp_id)
|
|
if parent is None:
|
|
parent = self.sox_client.load(comp_id)
|
|
if parent is None:
|
|
return []
|
|
uncached = [cid for cid in parent.children
|
|
if self.sox_client.cache_get(cid) is None]
|
|
if uncached:
|
|
self.sox_client.load_ids(uncached, checked=False)
|
|
return [self.sox_client.cache_get(cid)
|
|
for cid in parent.children
|
|
if self.sox_client.cache_get(cid) is not None]
|
|
|
|
# ── Wiresheet Open / Close ──────────────────────────────────
|
|
|
|
def open_wiresheet(self, folder_comp_id):
|
|
"""Full wiresheet load: children, links, remote comps, config+runtime, subscribe.
|
|
|
|
Returns list of SoxComponent for the folder's children.
|
|
"""
|
|
children = self.load_children(folder_comp_id)
|
|
if not children:
|
|
return []
|
|
|
|
all_link_remote_ids = set()
|
|
for child in children:
|
|
links = self.sox_client.read_links(child.id)
|
|
child.links = links
|
|
for lnk in links:
|
|
all_link_remote_ids.add(lnk.from_comp_id)
|
|
all_link_remote_ids.add(lnk.to_comp_id)
|
|
|
|
child_ids = {c.id for c in children}
|
|
child_ids.add(folder_comp_id)
|
|
remote_ids = [rid for rid in all_link_remote_ids
|
|
if rid not in child_ids
|
|
and self.sox_client.cache_get(rid) is None]
|
|
if remote_ids:
|
|
self.sox_client.load_ids(remote_ids, checked=False)
|
|
|
|
self.sox_client.update(children, CONFIG | RUNTIME)
|
|
|
|
folder_comp = self.sox_client.cache_get(folder_comp_id)
|
|
to_subscribe = [folder_comp] + children if folder_comp else children
|
|
to_subscribe = [c for c in to_subscribe if c is not None]
|
|
try:
|
|
self.sox_client.subscribe(to_subscribe, ALL)
|
|
except Exception:
|
|
pass
|
|
|
|
self._current_subs = {c.id for c in to_subscribe}
|
|
return children
|
|
|
|
def close_wiresheet(self):
|
|
"""Unsubscribe from previously subscribed components."""
|
|
if not self._current_subs or not self.is_connected:
|
|
self._current_subs.clear()
|
|
return
|
|
comps = [self.sox_client.cache_get(cid)
|
|
for cid in self._current_subs]
|
|
comps = [c for c in comps if c is not None]
|
|
if comps:
|
|
try:
|
|
self.sox_client.unsubscribe(comps, ALL)
|
|
except Exception:
|
|
pass
|
|
self._current_subs.clear()
|
|
|
|
# ── Post-add / Post-delete bookkeeping ─────────────────────
|
|
|
|
def after_add(self, comp_id):
|
|
"""Subscribes a newly added component and reads its initial values."""
|
|
comp = self.sox_client.cache_get(comp_id)
|
|
if comp is None or not self.is_connected:
|
|
return
|
|
try:
|
|
self.sox_client.update([comp], CONFIG | RUNTIME)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
self.sox_client.subscribe([comp], ALL)
|
|
self._current_subs.add(comp_id)
|
|
except Exception:
|
|
pass
|
|
|
|
def after_delete(self, comp_id, descendant_ids=None):
|
|
"""Removes a deleted component (and its descendants) from the subscription set."""
|
|
self._current_subs.discard(comp_id)
|
|
if descendant_ids:
|
|
for did in descendant_ids:
|
|
self._current_subs.discard(did)
|
|
|
|
# ── DOM Synthesis ───────────────────────────────────────────
|
|
|
|
def build_dom_tree(self, root_comp):
|
|
"""Builds an ET.Element tree from cached SoxComponents, for the editor UI.
|
|
|
|
Only builds from what is currently in the cache (lazy-loaded components).
|
|
Returns a root <sedonaApp> element.
|
|
"""
|
|
sax_root = ET.Element("sedonaApp")
|
|
schema_elem = ET.SubElement(sax_root, "schema")
|
|
if self.sox_client and self.sox_client.schema_parts:
|
|
for part in self.sox_client.schema_parts:
|
|
ET.SubElement(schema_elem, "kit",
|
|
name=part.name, checksum=str(part.checksum))
|
|
|
|
app_elem = ET.SubElement(sax_root, "app")
|
|
if root_comp is not None:
|
|
# Build children of root directly under <app>, matching .sax format
|
|
# where <app> IS the root container (not a <comp>)
|
|
for child_id in root_comp.children:
|
|
child_comp = self.sox_client.cache_get(child_id)
|
|
if child_comp is not None:
|
|
self._build_comp_element(app_elem, child_comp)
|
|
else:
|
|
ET.SubElement(app_elem, "comp",
|
|
name="...", id=str(child_id), type="")
|
|
|
|
return sax_root
|
|
|
|
def _build_comp_element(self, parent_elem, sox_comp):
|
|
type_qname = sox_comp.type_name
|
|
attribs = {
|
|
"name": sox_comp.name,
|
|
"id": str(sox_comp.id),
|
|
"type": type_qname,
|
|
}
|
|
comp_elem = ET.SubElement(parent_elem, "comp", **attribs)
|
|
|
|
slots = self.sox_client.slot_registry.get(
|
|
(sox_comp.kit_id, sox_comp.type_id))
|
|
if slots:
|
|
for slot in slots:
|
|
if slot.get("is_action"):
|
|
continue
|
|
name = slot["name"]
|
|
if name in sox_comp.values:
|
|
val = sox_comp.values[name]
|
|
if isinstance(val, float):
|
|
val_str = _fmt_float(val, _slot_precision(type_qname, name))
|
|
else:
|
|
val_str = str(val)
|
|
ET.SubElement(comp_elem, "prop",
|
|
name=name, val=val_str)
|
|
|
|
for child_id in sox_comp.children:
|
|
child_comp = self.sox_client.cache_get(child_id)
|
|
if child_comp is not None:
|
|
self._build_comp_element(comp_elem, child_comp)
|
|
else:
|
|
# Placeholder so the tree shows a + expand arrow for uncached children
|
|
ET.SubElement(comp_elem, "comp",
|
|
name="...", id=str(child_id), type="")
|
|
|
|
return comp_elem
|
|
|
|
def build_links_elements(self, sax_root):
|
|
"""Adds <link> elements under a <links> container, matching .sax structure."""
|
|
links_elem = sax_root.find("links")
|
|
if links_elem is None:
|
|
links_elem = ET.SubElement(sax_root, "links")
|
|
|
|
seen = set()
|
|
for c in self.sox_client.cache:
|
|
if c is None:
|
|
continue
|
|
for lnk in c.links:
|
|
link_key = (lnk.from_comp_id, lnk.from_slot_id,
|
|
lnk.to_comp_id, lnk.to_slot_id)
|
|
if link_key in seen:
|
|
continue
|
|
seen.add(link_key)
|
|
|
|
from_comp = self.sox_client.cache_get(lnk.from_comp_id)
|
|
to_comp = self.sox_client.cache_get(lnk.to_comp_id)
|
|
if from_comp is None or to_comp is None:
|
|
continue
|
|
|
|
from_slot_name = self._slot_name_by_id(
|
|
from_comp.kit_id, from_comp.type_id, lnk.from_slot_id)
|
|
to_slot_name = self._slot_name_by_id(
|
|
to_comp.kit_id, to_comp.type_id, lnk.to_slot_id)
|
|
|
|
from_path = self._comp_path(lnk.from_comp_id)
|
|
to_path = self._comp_path(lnk.to_comp_id)
|
|
|
|
if from_path and to_path and from_slot_name and to_slot_name:
|
|
ET.SubElement(links_elem, "link",
|
|
**{"from": f"{from_path}.{from_slot_name}",
|
|
"to": f"{to_path}.{to_slot_name}"})
|
|
|
|
def _slot_name_by_id(self, kit_id, type_id_val, slot_id):
|
|
# SOX links use absolute slot IDs across the full inheritance chain.
|
|
# final_components_registry has correctly resolved absolute IDs;
|
|
# the per-type slot_registry only stores local (per-type) IDs
|
|
# which are wrong for types with intermediate parents.
|
|
type_name = self.sox_client.type_name_map.get((kit_id, type_id_val))
|
|
if type_name and type_name in app_state.final_components_registry:
|
|
for s in app_state.final_components_registry[type_name]:
|
|
if s["slot_id"] == slot_id:
|
|
return s["name"]
|
|
# Fallback to slot_registry (local IDs — wrong for inherited types,
|
|
# but keeps things working when the schema registry is not yet loaded)
|
|
slots = self.sox_client.slot_registry.get((kit_id, type_id_val))
|
|
if slots:
|
|
for s in slots:
|
|
if s["id"] == slot_id:
|
|
return s["name"]
|
|
return None
|
|
|
|
def _comp_path(self, comp_id):
|
|
parts = []
|
|
cid = comp_id
|
|
while cid >= 0:
|
|
c = self.sox_client.cache_get(cid)
|
|
if c is None:
|
|
break
|
|
if c.id == 0:
|
|
break
|
|
parts.append(c.name)
|
|
cid = c.parent
|
|
if not parts:
|
|
return None
|
|
parts.reverse()
|
|
return "/" + "/".join(parts)
|
|
|
|
# ── Update DOM from COV ─────────────────────────────────────
|
|
|
|
def update_element_from_cache(self, comp_id, sax_root):
|
|
"""Updates a single comp element in the DOM from cached SoxComponent values."""
|
|
sox_comp = self.sox_client.cache_get(comp_id)
|
|
if sox_comp is None:
|
|
return None
|
|
|
|
comp_elem = self._find_element_by_id(sax_root, comp_id)
|
|
if comp_elem is None:
|
|
return None
|
|
|
|
type_name = sox_comp.type_name
|
|
slots = self.sox_client.slot_registry.get(
|
|
(sox_comp.kit_id, sox_comp.type_id))
|
|
if slots:
|
|
existing_props = {p.get("name"): p for p in comp_elem.findall("prop")}
|
|
for slot in slots:
|
|
if slot.get("is_action"):
|
|
continue
|
|
name = slot["name"]
|
|
if name in sox_comp.values:
|
|
raw = sox_comp.values[name]
|
|
if isinstance(raw, float):
|
|
val = _fmt_float(raw, _slot_precision(type_name, name))
|
|
else:
|
|
val = str(raw)
|
|
if name in existing_props:
|
|
existing_props[name].set("val", val)
|
|
else:
|
|
ET.SubElement(comp_elem, "prop", name=name, val=val)
|
|
return comp_elem
|
|
|
|
@staticmethod
|
|
def _find_element_by_id(root, comp_id):
|
|
target = str(comp_id)
|
|
for elem in root.iter("comp"):
|
|
if elem.get("id") == target:
|
|
return elem
|
|
return None
|
|
|
|
# ── Event queue (thread-safe) ───────────────────────────────
|
|
|
|
def _on_sox_event(self, comp_id, what_char):
|
|
self._event_queue.put((comp_id, what_char))
|
|
|
|
def poll_events(self):
|
|
"""Drains the event queue. Returns list of (comp_id, what_char) tuples."""
|
|
events = []
|
|
try:
|
|
while True:
|
|
events.append(self._event_queue.get_nowait())
|
|
except queue.Empty:
|
|
pass
|
|
return events
|