Files
sedona_editor/sox.py
T
ArdaandClaude Opus 4.6 dafebcebda Live Editor: add DASP/Sox device connection with full CRUD and COV
Port the DASP transport and Sox protocol stack from sedonaOntrol Java
to Python, enabling live device editing alongside the existing offline
SAX editor. Lazy-loads components on navigation, caches them, and
subscribes to COV events for the active wiresheet.

New files: dasp.py, sox.py, device_session.py, connect_dialog.py,
live_hooks.py. Modified: app_state.py (live-mode state), main.py
(Device menu, COV polling, lazy tree expand), properties.py (Sox
write-through), sax_file.py (structural mutations route through
live_hooks), schema.py (find_manifest helper).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-11 23:31:36 +03:00

1030 lines
35 KiB
Python

"""
Sox (Sedona Object Exchange) protocol layer.
Ported from sedonaOntrol/src/sedona/sox/ Java sources.
Provides component tree access, property read/write,
subscribe/unsubscribe, and structural mutations over DASP.
"""
import struct
import threading
import time
from dasp import DaspSocket, DaspException
# ────────────────────────────────────────────────────────────────
# Constants
# ────────────────────────────────────────────────────────────────
TREE = 0x01
CONFIG = 0x02
RUNTIME = 0x04
LINKS = 0x08
ALL = TREE | CONFIG | RUNTIME | LINKS
# Sedona type ids (matches Type.java constants)
BOOL_ID = 0
BYTE_ID = 1
SHORT_ID = 2
INT_ID = 3
LONG_ID = 4
FLOAT_ID = 5
DOUBLE_ID = 6
BUF_ID = 7
STR_ID = 8
class SoxException(Exception):
pass
# ────────────────────────────────────────────────────────────────
# SoxMsg — cursor-based binary buffer for Sox payloads
# ────────────────────────────────────────────────────────────────
class SoxMsg:
"""Big-endian binary buffer with cursor, mirroring sedona.Buf."""
def __init__(self, data=None, capacity=512):
if data is not None:
self._buf = bytearray(data)
self._size = len(data)
else:
self._buf = bytearray(capacity)
self._size = 0
self._pos = 0
# ── Read ────────────────────────────────────────────────────
def u1(self):
v = self._buf[self._pos]
self._pos += 1
return v
def u2(self):
v = (self._buf[self._pos] << 8) | self._buf[self._pos + 1]
self._pos += 2
return v
def i4(self):
v, = struct.unpack_from(">i", self._buf, self._pos)
self._pos += 4
return v
def i8(self):
v, = struct.unpack_from(">q", self._buf, self._pos)
self._pos += 8
return v
def f4(self):
v, = struct.unpack_from(">f", self._buf, self._pos)
self._pos += 4
return v
def f8(self):
v, = struct.unpack_from(">d", self._buf, self._pos)
self._pos += 8
return v
def str(self):
start = self._pos
while self._buf[self._pos] != 0:
self._pos += 1
s = self._buf[start:self._pos].decode("utf-8", errors="replace")
self._pos += 1 # null terminator
return s
def read_bytes(self, n):
data = bytes(self._buf[self._pos:self._pos + n])
self._pos += n
return data
@property
def remaining(self):
return self._size - self._pos
# ── Write ───────────────────────────────────────────────────
def _grow(self, need):
end = self._pos + need
if end > len(self._buf):
new_cap = max(end, len(self._buf) * 2)
self._buf.extend(b'\x00' * (new_cap - len(self._buf)))
if end > self._size:
self._size = end
def write_u1(self, v):
self._grow(1)
self._buf[self._pos] = v & 0xFF
self._pos += 1
def write_u2(self, v):
self._grow(2)
self._buf[self._pos] = (v >> 8) & 0xFF
self._buf[self._pos + 1] = v & 0xFF
self._pos += 2
def write_i4(self, v):
self._grow(4)
struct.pack_into(">i", self._buf, self._pos, v)
self._pos += 4
def write_i8(self, v):
self._grow(8)
struct.pack_into(">q", self._buf, self._pos, v)
self._pos += 8
def write_f4(self, v):
self._grow(4)
struct.pack_into(">f", self._buf, self._pos, v)
self._pos += 4
def write_f8(self, v):
self._grow(8)
struct.pack_into(">d", self._buf, self._pos, v)
self._pos += 8
def write_str(self, s):
encoded = s.encode("utf-8")
self._grow(len(encoded) + 1)
self._buf[self._pos:self._pos + len(encoded)] = encoded
self._pos += len(encoded)
self._buf[self._pos] = 0
self._pos += 1
def write_bytes(self, data):
self._grow(len(data))
self._buf[self._pos:self._pos + len(data)] = data
self._pos += len(data)
# ── Payload ─────────────────────────────────────────────────
def to_bytes(self):
return bytes(self._buf[:self._size])
@property
def command(self):
return self._buf[0]
@property
def reply_num(self):
return self._buf[1] & 0xFF
@reply_num.setter
def reply_num(self, v):
self._buf[1] = v & 0xFF
def check_response(self, expected_cmd):
cmd = self.u1()
_reply = self.u1()
if cmd == ord('!'):
cause = self.str()
raise SoxException(f"Request failed: {cause}")
if cmd != ord(expected_cmd):
raise SoxException(
f"Response {chr(cmd)}({cmd}) != expected {expected_cmd}({ord(expected_cmd)})")
# ────────────────────────────────────────────────────────────────
# Value codec — encode/decode Sedona property values
# ────────────────────────────────────────────────────────────────
def decode_value(msg, type_id):
if type_id == BOOL_ID:
return msg.u1() != 0
elif type_id == BYTE_ID:
return msg.u1()
elif type_id == SHORT_ID:
return msg.u2()
elif type_id == INT_ID:
return msg.i4()
elif type_id == LONG_ID:
return msg.i8()
elif type_id == FLOAT_ID:
return msg.f4()
elif type_id == DOUBLE_ID:
return msg.f8()
elif type_id == STR_ID:
size = msg.u2()
if size == 0:
return ""
chars = msg.read_bytes(size - 1).decode("utf-8", errors="replace")
msg.u1() # null terminator
return chars
elif type_id == BUF_ID:
size = msg.u2()
return msg.read_bytes(size)
else:
raise SoxException(f"Unknown type id: {type_id}")
def encode_value(msg, type_id, value):
if type_id == BOOL_ID:
msg.write_u1(1 if value else 0)
elif type_id == BYTE_ID:
msg.write_u1(value & 0xFF)
elif type_id == SHORT_ID:
msg.write_u2(value & 0xFFFF)
elif type_id == INT_ID:
msg.write_i4(value)
elif type_id == LONG_ID:
msg.write_i8(value)
elif type_id == FLOAT_ID:
msg.write_f4(value)
elif type_id == DOUBLE_ID:
msg.write_f8(value)
elif type_id == STR_ID:
encoded = value.encode("utf-8") if isinstance(value, str) else value
msg.write_u2(len(encoded) + 1)
msg.write_bytes(encoded)
msg.write_u1(0)
elif type_id == BUF_ID:
msg.write_u2(len(value))
msg.write_bytes(value)
else:
raise SoxException(f"Unknown type id: {type_id}")
TYPE_ID_BY_NAME = {
"bool": BOOL_ID, "byte": BYTE_ID, "short": SHORT_ID,
"int": INT_ID, "long": LONG_ID, "float": FLOAT_ID,
"double": DOUBLE_ID, "buf": BUF_ID, "Buf": BUF_ID,
}
def type_id_for_slot(slot_type_name):
if slot_type_name in TYPE_ID_BY_NAME:
return TYPE_ID_BY_NAME[slot_type_name]
return STR_ID # asStr slots
# ────────────────────────────────────────────────────────────────
# Link
# ────────────────────────────────────────────────────────────────
class Link:
__slots__ = ("from_comp_id", "from_slot_id", "to_comp_id", "to_slot_id")
def __init__(self, from_comp_id=0, from_slot_id=0, to_comp_id=0, to_slot_id=0):
self.from_comp_id = from_comp_id
self.from_slot_id = from_slot_id
self.to_comp_id = to_comp_id
self.to_slot_id = to_slot_id
def __repr__(self):
return (f"Link({self.from_comp_id}:{self.from_slot_id}"
f" -> {self.to_comp_id}:{self.to_slot_id})")
# ────────────────────────────────────────────────────────────────
# SoxComponent — cached remote component
# ────────────────────────────────────────────────────────────────
class SoxComponent:
def __init__(self, comp_id, kit_id=-1, type_id=-1, type_name=""):
self.id = comp_id
self.kit_id = kit_id
self.type_id = type_id
self.type_name = type_name # "kit::Type" qualified name
self.name = ""
self.parent = -1
self.children = [] # list of child comp ids
self.permissions = 0
self.subscription = 0 # bitmask of subscribed categories
self.links = [] # list of Link objects
self.values = {} # slot_name -> value (config + runtime)
def __repr__(self):
return f"SoxComponent(id={self.id}, name={self.name!r}, type={self.type_name})"
# ────────────────────────────────────────────────────────────────
# KitPart — kit name + checksum from schema read
# ────────────────────────────────────────────────────────────────
class KitPart:
__slots__ = ("name", "checksum")
def __init__(self, name, checksum):
self.name = name
self.checksum = checksum
def __repr__(self):
return f"{self.name}-{self.checksum:08x}"
class VersionInfo:
def __init__(self, platform_id, scode_flags, kit_versions, props=None):
self.platform_id = platform_id
self.scode_flags = scode_flags
self.kit_versions = kit_versions # list of (kit_name, checksum, version_str)
self.props = props or {} # extra props like "soxVer"
# ────────────────────────────────────────────────────────────────
# SoxClient — Sox protocol client with component cache
# ────────────────────────────────────────────────────────────────
class SoxClient:
"""
Sox client with component cache, request/response correlation,
batch subscribe, and COV event dispatch.
"""
def __init__(self, host, port, user, password, **dasp_options):
self.host = host
self.port = port
self.user = user
self.password = password
self.dasp_options = dasp_options
self._dasp_socket = None
self._session = None
self._receiver = None
self._lock = threading.Lock()
# request/response correlation
self._requests = None # list[SoxMsg] current batch
self._responses = None # list[SoxMsg|None] current batch
self._req_event = threading.Event()
# cache
self.cache = [None] * 1024 # SoxComponent[] indexed by comp_id
self.all_tree_events = False
# schema
self.schema_parts = None # list[KitPart]
self.version_info = None # VersionInfo
self.schema = None # will hold schema reference from editor
# slot registry — set externally after readSchema
# maps (kit_id, type_id) -> list of slot dicts
# each slot dict: {name, id, type_name, is_config, is_runtime, is_action, is_operator, is_as_str}
self.slot_registry = {}
# event callback: called on main thread via queue
self.on_event = None # fn(comp_id, what_char)
# ── Connect / Disconnect ────────────────────────────────────
def connect(self):
self._dasp_socket = DaspSocket()
self._session = self._dasp_socket.connect(
self.host, self.port, self.user, self.password,
**self.dasp_options)
self._receiver = _SoxReceiver(self)
self._receiver.start()
def close(self):
if self._receiver:
self._receiver.kill()
self._receiver = None
if self._session and not self._session.is_closed:
try:
self._session.close()
except Exception:
pass
if self._dasp_socket:
try:
self._dasp_socket.close()
except Exception:
pass
self._session = None
self._dasp_socket = None
self.cache = [None] * 1024
self.all_tree_events = False
@property
def is_closed(self):
return self._session is None or self._session.is_closed
# ── Read Schema ─────────────────────────────────────────────
def read_schema(self):
if self.schema_parts is not None:
return self.schema_parts
req = _prepare_request(ord('v'))
res = self.request(req)
res.check_response('V')
num = res.u1()
parts = []
for _ in range(num):
name = res.str()
checksum = res.i4()
parts.append(KitPart(name, checksum))
self.schema_parts = parts
return parts
def read_version(self):
if self.version_info is not None:
return self.version_info
parts = self.read_schema()
req = _prepare_request(ord('y'))
res = self.request(req)
res.check_response('Y')
platform_id = res.str()
scode_flags = res.u1()
kit_versions = []
for p in parts:
ver = res.str()
kit_versions.append((p.name, p.checksum, ver))
vi = VersionInfo(platform_id, scode_flags, kit_versions)
prop_num = res.u1()
for _ in range(prop_num):
k = res.str()
v = res.str()
vi.props[k] = v
self.version_info = vi
return vi
@property
def sox_version(self):
try:
vi = self.read_version()
return vi.props.get("soxVer")
except Exception:
return None
# ── Load Component (Tree) ───────────────────────────────────
def load_app(self):
return self.load(0)
def load(self, comp_id):
if comp_id < 0:
return None
cached = self.cache_get(comp_id)
if cached is not None:
return cached
results = self.load_ids([comp_id])
return results[0] if results else None
def load_ids(self, ids, checked=True):
self.read_schema()
results = [None] * len(ids)
reqs = []
req_indices = []
for i, cid in enumerate(ids):
c = self.cache_get(cid)
if c is not None:
results[i] = c
else:
reqs.append(_make_update_req(cid, ord('t')))
req_indices.append(i)
if not reqs:
return results
responses = self.requests(reqs)
for res in responses:
res.check_response('C')
self._apply_to_cache(res)
for i, cid in enumerate(ids):
results[i] = self.cache_get(cid)
if checked and results[i] is None:
raise SoxException(f"Failed to load component {cid}")
return results
# ── Update (config / runtime / links) ───────────────────────
def update(self, comps, mask):
reqs = []
for c in comps:
if mask & TREE:
reqs.append(_make_update_req(c.id, ord('t')))
if mask & CONFIG:
reqs.append(_make_update_req(c.id, ord('c')))
if mask & RUNTIME:
reqs.append(_make_update_req(c.id, ord('r')))
if mask & LINKS:
reqs.append(_make_update_req(c.id, ord('l')))
if not reqs:
return
responses = self.requests(reqs)
for res in responses:
res.check_response('C')
self._apply_to_cache(res)
# ── Subscribe / Unsubscribe ─────────────────────────────────
def subscribe(self, comps, mask):
if self.sox_version is not None:
self._batch_subscribe(comps, mask)
else:
self._do_subscribe(comps, mask)
def _batch_subscribe(self, comps, mask):
to_sub = [c for c in comps if (c.subscription & mask) != mask]
if not to_sub:
return
req = _prepare_request(ord('s'))
req.write_u1(mask)
req.write_u1(len(to_sub))
for c in to_sub:
req.write_u2(c.id)
res = self.request(req)
res.check_response('S')
for c in to_sub:
c.subscription |= mask
def _do_subscribe(self, comps, mask):
reqs = []
for c in comps:
cur = c.subscription
need = mask & ~cur
if need & TREE:
reqs.append(_make_subscribe_req(c.id, ord('t')))
if need & CONFIG:
reqs.append(_make_subscribe_req(c.id, ord('c')))
if need & RUNTIME:
reqs.append(_make_subscribe_req(c.id, ord('r')))
if need & LINKS:
reqs.append(_make_subscribe_req(c.id, ord('l')))
if not reqs:
return
responses = self.requests(reqs)
for res in responses:
res.check_response('S')
self._apply_to_cache(res)
for c in comps:
c.subscription |= mask
def subscribe_all_tree_events(self):
if self.sox_version is not None:
req = _prepare_request(ord('s'))
req.write_u1(0xFF)
req.write_u1(0)
self.request(req)
else:
self.request(_make_subscribe_req(0, ord('a')))
self.all_tree_events = True
for c in self.cache:
if c is not None:
c.subscription |= TREE
def unsubscribe(self, comps, mask):
if self.is_closed:
return
if self.all_tree_events:
mask &= ~TREE
to_unsub = [c for c in comps if (c.subscription & mask) != 0]
if not to_unsub:
return
if self.sox_version is not None:
self._batch_unsubscribe(to_unsub, mask)
else:
self._do_unsubscribe(to_unsub, mask)
for c in comps:
c.subscription &= ~mask
def _batch_unsubscribe(self, comps, mask):
req = _prepare_request(ord('u'))
req.write_u1(mask)
req.write_u1(len(comps))
for c in comps:
req.write_u2(c.id)
self.request(req)
def _do_unsubscribe(self, comps, mask):
reqs = []
for c in comps:
reqs.append(_make_unsubscribe_req(c.id, mask))
self.requests(reqs)
# ── Read Property ───────────────────────────────────────────
def read_prop(self, comp_id, slot_id, type_id):
req = _prepare_request(ord('r'))
req.write_u2(comp_id)
req.write_u1(slot_id)
res = self.request(req)
res.check_response('R')
res.u2() # comp_id
res.u1() # slot_id
res_type_id = res.u1()
return decode_value(res, res_type_id)
# ── Write Property ──────────────────────────────────────────
def write_prop(self, comp_id, slot_id, type_id, value):
req = _prepare_request(ord('w'))
req.write_u2(comp_id)
req.write_u1(slot_id)
encode_value(req, type_id, value)
res = self.request(req)
res.check_response('W')
# ── Invoke Action ───────────────────────────────────────────
def invoke(self, comp_id, slot_id, type_id=None, arg=None):
req = _prepare_request(ord('i'))
req.write_u2(comp_id)
req.write_u1(slot_id)
if arg is not None and type_id is not None:
encode_value(req, type_id, arg)
res = self.request(req)
res.check_response('I')
# ── Add Component ───────────────────────────────────────────
def add(self, parent_id, kit_id, type_id_val, name, config_values=None):
req = _prepare_request(ord('a'))
req.write_u2(parent_id)
req.write_u1(kit_id)
req.write_u1(type_id_val)
req.write_str(name)
if config_values:
for tid, val in config_values:
encode_value(req, tid, val)
res = self.request(req)
res.check_response('A')
new_id = res.u2()
comp = SoxComponent(new_id, kit_id, type_id_val)
comp.name = name
comp.parent = parent_id
comp.children = []
self.cache_add(comp)
parent = self.cache_get(parent_id)
if parent is not None and new_id not in parent.children:
parent.children.append(new_id)
return comp
# ── Delete Component ────────────────────────────────────────
def delete(self, comp_id):
req = _prepare_request(ord('d'))
req.write_u2(comp_id)
res = self.request(req)
res.check_response('D')
comp = self.cache_get(comp_id)
if comp is not None:
parent = self.cache_get(comp.parent)
if parent is not None and comp_id in parent.children:
parent.children.remove(comp_id)
self.cache_remove(comp)
# ── Rename Component ────────────────────────────────────────
def rename(self, comp_id, new_name):
req = _prepare_request(ord('n'))
req.write_u2(comp_id)
req.write_str(new_name)
res = self.request(req)
res.check_response('N')
comp = self.cache_get(comp_id)
if comp is not None:
comp.name = new_name
# ── Reorder Children ────────────────────────────────────────
def reorder(self, comp_id, child_ids):
req = _prepare_request(ord('o'))
req.write_u2(comp_id)
req.write_u1(len(child_ids))
for cid in child_ids:
req.write_u2(cid)
res = self.request(req)
res.check_response('O')
comp = self.cache_get(comp_id)
if comp is not None:
comp.children = list(child_ids)
# ── Link / Unlink ───────────────────────────────────────────
def link(self, from_comp_id, from_slot_id, to_comp_id, to_slot_id):
self._link_cmd(ord('a'), from_comp_id, from_slot_id, to_comp_id, to_slot_id)
def unlink(self, from_comp_id, from_slot_id, to_comp_id, to_slot_id):
self._link_cmd(ord('d'), from_comp_id, from_slot_id, to_comp_id, to_slot_id)
def _link_cmd(self, subcmd, from_comp_id, from_slot_id, to_comp_id, to_slot_id):
req = _prepare_request(ord('l'))
req.write_u1(subcmd)
req.write_u2(from_comp_id)
req.write_u1(from_slot_id)
req.write_u2(to_comp_id)
req.write_u1(to_slot_id)
res = self.request(req)
res.check_response('L')
# ── Links snapshot ──────────────────────────────────────────
def read_links(self, comp_id):
req = _prepare_request(ord('c'))
req.write_u2(comp_id)
req.write_u1(ord('l'))
res = self.request(req)
res.check_response('C')
res_comp_id = res.u2()
what = res.u1()
links = []
while True:
from_comp = res.u2()
if from_comp == 0xFFFF:
break
from_slot = res.u1()
to_comp = res.u2()
to_slot = res.u1()
links.append(Link(from_comp, from_slot, to_comp, to_slot))
return links
# ── Cache Management ────────────────────────────────────────
def cache_get(self, comp_id):
if 0 <= comp_id < len(self.cache):
return self.cache[comp_id]
return None
def cache_add(self, comp):
if comp.id >= len(self.cache):
new_size = max(len(self.cache) * 2, comp.id + 32)
self.cache.extend([None] * (new_size - len(self.cache)))
self.cache[comp.id] = comp
def cache_remove(self, comp):
for child_id in list(comp.children):
child = self.cache_get(child_id)
if child is not None:
self.cache_remove(child)
if 0 <= comp.id < len(self.cache):
self.cache[comp.id] = None
# ── Apply to Cache ──────────────────────────────────────────
def _apply_to_cache(self, msg, is_event=False):
comp_id = msg.u2()
what = msg.u1()
cached = self.cache_get(comp_id)
if cached is None:
if what == ord('t'):
cached = self._apply_tree_new(msg, comp_id)
self.cache_add(cached)
else:
return
else:
if what == ord('t'):
self._apply_tree_cached(msg, cached)
elif what in (ord('c'), ord('C')):
self._apply_props(msg, cached, 'c', what == ord('C'))
elif what in (ord('r'), ord('R')):
self._apply_props(msg, cached, 'r', what == ord('R'))
elif what == ord('l'):
self._apply_links(msg, cached)
if is_event and cached is not None:
mask = _what_to_mask(what)
if mask and not (cached.subscription & mask):
cached.subscription |= mask
if self.on_event:
self.on_event(comp_id, chr(what))
def _apply_tree_new(self, msg, comp_id):
kit_id = msg.u1()
type_id_val = msg.u1()
type_name = self._resolve_type_name(kit_id, type_id_val)
comp = SoxComponent(comp_id, kit_id, type_id_val, type_name)
self._apply_tree_body(msg, comp)
return comp
def _apply_tree_cached(self, msg, comp):
_kit_id = msg.u1()
_type_id_val = msg.u1()
self._apply_tree_body(msg, comp)
def _apply_tree_body(self, msg, comp):
comp.name = msg.str()
comp.parent = msg.u2()
comp.permissions = msg.u1()
num_children = msg.u1()
new_children = [msg.u2() for _ in range(num_children)]
old_set = set(comp.children) if comp.children else set()
new_set = set(new_children)
for removed_id in old_set - new_set:
removed = self.cache_get(removed_id)
if removed is not None:
self.cache_remove(removed)
comp.children = new_children
def _apply_props(self, msg, comp, what_char, operator_only):
slots = self._get_slots(comp.kit_id, comp.type_id)
if slots is None:
return
for slot in slots:
if not slot.get("is_prop", True):
continue
if slot.get("is_action"):
continue
if operator_only and not slot.get("is_operator"):
continue
is_config = slot.get("is_config", True)
is_runtime = slot.get("is_runtime", False)
if what_char == 'c' and is_runtime:
continue
if what_char == 'r' and is_config:
continue
tid = type_id_for_slot(slot.get("type_name", "int"))
try:
val = decode_value(msg, tid)
comp.values[slot["name"]] = val
except Exception:
break
def _apply_links(self, msg, comp):
links = []
while msg.remaining >= 6:
from_comp = msg.u2()
if from_comp == 0xFFFF:
break
from_slot = msg.u1()
to_comp = msg.u2()
to_slot = msg.u1()
links.append(Link(from_comp, from_slot, to_comp, to_slot))
comp.links = links
def _resolve_type_name(self, kit_id, type_id_val):
if self.schema_parts and 0 <= kit_id < len(self.schema_parts):
kit_name = self.schema_parts[kit_id].name
else:
kit_name = f"kit{kit_id}"
return f"{kit_name}::type{type_id_val}"
def _get_slots(self, kit_id, type_id_val):
return self.slot_registry.get((kit_id, type_id_val))
# ── Networking ──────────────────────────────────────────────
def request(self, req):
results = self.requests([req])
return results[0]
def requests(self, reqs):
if self.is_closed:
raise SoxException("SoxClient closed")
if len(reqs) > 255:
return self._chunk_requests(reqs)
with self._lock:
for i, req in enumerate(reqs):
req.reply_num = i
self._responses = [None] * len(reqs)
self._requests = reqs
self._req_event.clear()
for req in reqs:
self._session.send(req.to_bytes())
deadline = time.monotonic() + 30.0
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
break
if all(r is not None for r in self._responses):
res = list(self._responses)
self._requests = None
self._responses = None
return res
self._req_event.wait(timeout=min(remaining, 0.5))
self._req_event.clear()
if self.is_closed:
raise SoxException("Session closed while waiting for response")
self._requests = None
self._responses = None
raise SoxException("Request timeout")
def _chunk_requests(self, reqs):
results = []
for i in range(0, len(reqs), 255):
chunk = reqs[i:i + 255]
results.extend(self.requests(chunk))
return results
# called by receiver thread
def _receive_response(self, msg):
reqs = self._requests
resps = self._responses
if reqs is None or resps is None:
return
cmd = msg.command
reply_num = msg.reply_num
index = reply_num & 0xFF
if index < 0 or index >= len(reqs):
return
req_cmd = reqs[index].command
if cmd != ord('!') and cmd != (req_cmd & ~0x20):
return
resps[index] = msg
self._req_event.set()
# called by receiver thread for events
def _receive_event(self, msg):
_cmd = msg.u1()
_reply = msg.u1()
try:
self._apply_to_cache(msg, is_event=True)
except Exception:
pass
# ────────────────────────────────────────────────────────────────
# SoxReceiver — daemon thread
# ────────────────────────────────────────────────────────────────
class _SoxReceiver(threading.Thread):
def __init__(self, client):
super().__init__(name="SoxReceiver", daemon=True)
self._client = client
self._alive = True
def kill(self):
self._alive = False
def run(self):
client = self._client
while self._alive and not client.is_closed:
try:
dasp_msg = client._session.receive(timeout=1.0)
if dasp_msg is None:
continue
payload = dasp_msg.payload
if not payload or len(payload) < 2:
continue
msg = SoxMsg(payload)
cmd = payload[0]
if cmd == ord('e'):
client._receive_event(msg)
elif cmd == ord('k') or cmd == ord('z'):
pass # file transfer — not needed for editor
else:
client._receive_response(msg)
except DaspException:
if self._alive:
break
except Exception:
if self._alive and not client.is_closed:
continue
# ────────────────────────────────────────────────────────────────
# Message factories
# ────────────────────────────────────────────────────────────────
def _prepare_request(cmd, reply_num=0xFF):
msg = SoxMsg()
msg.write_u1(cmd)
msg.write_u1(reply_num)
return msg
def _make_update_req(comp_id, what):
req = _prepare_request(ord('c'))
req.write_u2(comp_id)
req.write_u1(what)
return req
def _make_subscribe_req(comp_id, what):
req = _prepare_request(ord('s'))
req.write_u2(comp_id)
req.write_u1(what)
return req
def _make_unsubscribe_req(comp_id, mask):
req = _prepare_request(ord('u'))
req.write_u2(comp_id)
req.write_u1(mask)
return req
def _what_to_mask(what):
if what == ord('t'):
return TREE
if what in (ord('c'), ord('C')):
return CONFIG
if what in (ord('r'), ord('R')):
return RUNTIME
if what == ord('l'):
return LINKS
return 0