ArdaandClaude Opus 5.5
cb0196caa7
Saves the app on the device, reads app.sab and fills the Sox cache with the tree, config and links in one go: 8 s instead of 116 s for 230 components on a P-ION, 0.9 s instead of 16 s for 194 on an Orion. Sox keeps the cache current from there as before. On any failure, or when app.sab does not match the running tree, it falls back to lazy loading. A sheet's first open reads only runtime values; a tree expand with nothing to load no longer rebuilds the DOM. bench_sab_seed.py times both ways against a device. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
618 lines
25 KiB
Python
618 lines
25 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
|
|
|
|
|
|
def _as_sox_value(slot, val):
|
|
"""An app.sab config value as a Sox config read leaves it in the cache."""
|
|
if "s" in slot.flags:
|
|
return val
|
|
if slot.type in ("sys::Buf", "buf"):
|
|
# Sox reads every Buf as a Str: its bytes but the last, as text.
|
|
return val[:-1].decode("utf-8", errors="replace") if val else ""
|
|
if slot.type == "bool":
|
|
return val is not False # Sox reads a bool as u1 != 0, so null is true
|
|
return val
|
|
|
|
|
|
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
|
|
# Components whose config and links came from app.sab and have not been read
|
|
# since: their first sheet open skips reading those again.
|
|
self._seeded = set()
|
|
self.seed_info = "" # how the cache was filled, for the status bar
|
|
|
|
# ── 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.checksum_resolver = schema.resolve_truncated_checksum
|
|
self.sox_client.connect()
|
|
|
|
parts = self.sox_client.read_schema()
|
|
vi = self.sox_client.read_version()
|
|
|
|
self._load_slot_registry(parts)
|
|
|
|
self.sox_client.load(0)
|
|
try:
|
|
count = self.seed_from_sab()
|
|
self.seed_info = f"app read from app.sab: {count} components"
|
|
except Exception as e:
|
|
self.seed_info = f"app.sab not used: {e}"
|
|
|
|
root_comp = self.sox_client.cache_get(0)
|
|
if root_comp and root_comp.children:
|
|
children = self.load_children(root_comp.id)
|
|
if children:
|
|
self.sox_client.update(children, self._values_to_read(children))
|
|
return root_comp
|
|
|
|
def disconnect(self, unsubscribe=True):
|
|
"""Closes the session. unsubscribe=False skips telling the device first -- for a
|
|
device going down, which would leave the request to time out."""
|
|
if self.sox_client is None:
|
|
return
|
|
try:
|
|
if unsubscribe and 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 = {}
|
|
own_slots = {} # "kit::Type" -> declared slots
|
|
bases = {} # "kit::Type" -> "kit::Base" or None
|
|
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
|
|
self._register_kit_types(registry, kit_id, kit_name, manifest,
|
|
own_slots, bases)
|
|
# Flatten each type's slots through its whole base chain (e.g.
|
|
# Component -> Service -> ontrolOx::Widget -> PushButtonImage), then
|
|
# renumber IDs to match the wire slot index (positional in the
|
|
# flattened inherited + own list — see sedona Type.java resolveSlots).
|
|
for key in registry:
|
|
qname = self.sox_client.type_name_map[key]
|
|
chain = []
|
|
seen = set()
|
|
t = qname
|
|
while t and t not in seen and t in own_slots:
|
|
seen.add(t)
|
|
chain.append(t)
|
|
t = bases.get(t)
|
|
flat = []
|
|
for t in reversed(chain):
|
|
flat.extend(dict(s) for s in own_slots[t])
|
|
for i, slot in enumerate(flat):
|
|
slot["id"] = i
|
|
registry[key] = flat
|
|
self.sox_client.slot_registry = registry
|
|
|
|
def _register_kit_types(self, registry, kit_id, kit_name, manifest,
|
|
own_slots, bases):
|
|
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}"
|
|
own_slots[full_name] = slots
|
|
bases[full_name] = type_elem.get("base") or None
|
|
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
|
|
|
|
# ── Seeding from app.sab ────────────────────────────────────
|
|
|
|
def seed_from_sab(self):
|
|
"""Fills the cache with the whole app — tree, config and links — from app.sab.
|
|
|
|
Reading component by component costs a round trip each, and a device answers one
|
|
request per pass of its loop: 230 components took 116 s on a P-ION, where saving
|
|
the app and reading app.sab took 8 s. The app is saved first, so app.sab holds
|
|
what is running. From here on Sox keeps the cache current, as it does for a
|
|
lazily loaded one; runtime values are still read when a sheet opens.
|
|
|
|
Nothing goes into the cache unless the whole file decodes and its tree matches the
|
|
device's, so on any failure the caller falls back to lazy loading. Returns the
|
|
number of components.
|
|
"""
|
|
import app_backup
|
|
import sab_app
|
|
from sox import Link
|
|
|
|
client = self.sox_client
|
|
root = client.cache_get(0) or client.load(0)
|
|
app_backup.save_app(client)
|
|
app = sab_app.SabApp(client.get_file(app_backup.APP_SAB))
|
|
|
|
# app.sab numbers kits by the app's schema and Sox by kits.scode's; the two can
|
|
# differ, so types are matched by name, and a kit only when its checksum agrees.
|
|
device_kits = {p.name: f"{p.checksum & 0xFFFFFFFF:08x}" for p in client.schema_parts}
|
|
for name, checksum in app.kit_parts:
|
|
if device_kits.get(name) != checksum:
|
|
raise ValueError(f"kit {name} in app.sab is not the device's")
|
|
type_key = {qname: key for key, qname in client.type_name_map.items()}
|
|
|
|
if [k["id"] for k in app.app["kids"]] != root.children:
|
|
raise ValueError("app.sab does not match the running app")
|
|
|
|
comps = {}
|
|
for c in app.comps.values():
|
|
key = type_key.get(c["type"])
|
|
if key is None:
|
|
raise ValueError(f"no manifest for {c['type']}")
|
|
comp = SoxComponent(c["id"], key[0], key[1], c["type"])
|
|
comp.name = c["name"]
|
|
comp.parent = c["parentId"]
|
|
comp.children = [k["id"] for k in c["kids"]]
|
|
comp.permissions = root.permissions
|
|
for slot, val in c["props"]:
|
|
comp.values[slot.name] = _as_sox_value(slot, val)
|
|
comps[comp.id] = comp
|
|
comps[0].subscription = root.subscription
|
|
|
|
# Sox reads a component's links both ways, into it and out of it.
|
|
for from_c, from_slot, to_c, to_slot in app.links:
|
|
link = Link(from_c["id"], from_slot, to_c["id"], to_slot)
|
|
comps[from_c["id"]].links.append(link)
|
|
if to_c["id"] != from_c["id"]:
|
|
comps[to_c["id"]].links.append(link)
|
|
|
|
for comp in comps.values():
|
|
client.cache_add(comp)
|
|
self._seeded = set(comps)
|
|
return len(comps)
|
|
|
|
def _values_to_read(self, comps):
|
|
"""What to read for `comps` on a sheet open: config too, unless app.sab gave it."""
|
|
if all(c.id in self._seeded for c in comps):
|
|
return RUNTIME
|
|
return CONFIG | RUNTIME
|
|
|
|
# ── 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]
|
|
|
|
def used_kit_names(self):
|
|
"""Every kit a component on the device belongs to, loading the whole tree to see.
|
|
|
|
The DOM only holds what has been opened, so it cannot answer this on its own. Only
|
|
the tree is read — no config or runtime.
|
|
|
|
Requests go one at a time. A small controller handles a single folder's burst fine,
|
|
but a back-to-back stream of them fills its receive buffer; it stops acking, and
|
|
DASP closes the whole session ("unacked send timeout"). Waiting for each reply
|
|
before sending the next keeps it at the pace the device sets. Anything already
|
|
cached costs nothing, so only the first call is slow.
|
|
"""
|
|
client = self.sox_client
|
|
if client is None:
|
|
return set()
|
|
|
|
pending = [0]
|
|
loaded = 0
|
|
while pending:
|
|
comp_id = pending.pop()
|
|
comp = client.cache_get(comp_id)
|
|
if comp is None:
|
|
comp = client.load_ids([comp_id], checked=False)[0]
|
|
loaded += 1
|
|
if loaded % 20 == 0:
|
|
app_state.set_status(f"Reading components from the device… {loaded}")
|
|
if app_state.status_label is not None:
|
|
app_state.status_label.update_idletasks()
|
|
if comp is not None:
|
|
pending.extend(comp.children)
|
|
|
|
parts = client.schema_parts or []
|
|
return {parts[c.kit_id].name for c in client.cache
|
|
if c is not None and 0 <= c.kit_id < len(parts)}
|
|
|
|
def find_component_of_type(self, type_name, slot_name):
|
|
"""A component of `type_name` with `slot_name` read, loading where services live.
|
|
|
|
Nothing is loaded until a folder is opened, so a platform service the user never
|
|
expanded to is not cached yet. It sits one or two levels down (/service/plat), and
|
|
those are loaded to find it rather than the whole app.
|
|
"""
|
|
client = self.sox_client
|
|
if client is None:
|
|
return None
|
|
|
|
def cached():
|
|
return next((c for c in client.cache
|
|
if c is not None and c.type_name == type_name), None)
|
|
|
|
comp = cached()
|
|
if comp is None:
|
|
for top in self.load_children(0):
|
|
if top.name == "service" or top.type_name == type_name:
|
|
for child in self.load_children(top.id):
|
|
if child.children and child.type_name != type_name:
|
|
self.load_children(child.id)
|
|
comp = cached()
|
|
if comp is not None and slot_name not in comp.values:
|
|
client.update([comp], CONFIG)
|
|
return comp
|
|
|
|
# ── 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:
|
|
if child.id not in self._seeded:
|
|
child.links = self.sox_client.read_links(child.id)
|
|
for lnk in child.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)
|
|
|
|
seeded = [c for c in children if c.id in self._seeded]
|
|
others = [c for c in children if c.id not in self._seeded]
|
|
if seeded:
|
|
self.sox_client.update(seeded, RUNTIME)
|
|
if others:
|
|
self.sox_client.update(others, CONFIG | RUNTIME)
|
|
# Subscribed from here on, so Sox keeps them current; a later open reads again.
|
|
self._seeded -= child_ids
|
|
|
|
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:
|
|
# Sox sends the checksum as a signed int; .sax files and manifest names
|
|
# spell it as 8 hex digits, and nothing matches unless this does too.
|
|
ET.SubElement(schema_elem, "kit",
|
|
name=part.name, checksum=f"{part.checksum & 0xFFFFFFFF:08x}")
|
|
|
|
app_elem = ET.SubElement(sax_root, "app")
|
|
if root_comp is not None:
|
|
app_elem.set("id", str(root_comp.id))
|
|
# The app is a component like any other, so the Properties view needs its
|
|
# type and slot values too. A .sax file leaves both implied; here they are not.
|
|
app_elem.set("type", root_comp.type_name)
|
|
self._add_prop_elements(app_elem, root_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)
|
|
self._add_prop_elements(comp_elem, sox_comp)
|
|
|
|
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 _add_prop_elements(self, comp_elem, sox_comp):
|
|
"""Writes a <prop> for every cached non-action slot value of `sox_comp`."""
|
|
slots = self.sox_client.slot_registry.get(
|
|
(sox_comp.kit_id, sox_comp.type_id))
|
|
if not slots:
|
|
return
|
|
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(sox_comp.type_name, name))
|
|
else:
|
|
val_str = str(val)
|
|
ET.SubElement(comp_elem, "prop", name=name, val=val_str)
|
|
|
|
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.
|
|
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 (flattened through the base chain, so
|
|
# its IDs are absolute too)
|
|
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)
|
|
app_elem = root.find("app")
|
|
if app_elem is not None and app_elem.get("id") == target:
|
|
return app_elem
|
|
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
|