Files
sedona_editor/sox.py
T
ArdaandClaude Opus 4.6 6cd030c7ea Fix live wiresheet config/runtime slot alignment and cap float display to 2 decimals
Slots with no flags (sensor, controlSignal, out_*) were not skipped
during config property reads, consuming bytes that the device never
sent and misaligning every subsequent config value (setpoint,
proportionalBand, etc.) into tiny denormalized floats like 2.36875e-41.

Changed the filter in _apply_props from "skip if runtime" to "skip if
not config" so only explicitly config-flagged slots are read from config
messages. Also switched iter("slot") to findall("slot") in the slot
registry builder to avoid picking up nested descendants.

Float display now defaults to 2 decimal places everywhere (wiresheet
and properties panel), honouring the precision facet when one exists.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 15:46:44 +03:00

1036 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 & 0xFFFFFFFF: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 = {}
# type name map — set externally by device_session
# maps (kit_id, type_id) -> "kit::TypeName" for proper type resolution
self.type_name_map = {}
# 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", False)
if what_char == 'c' and not is_config:
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):
resolved = self.type_name_map.get((kit_id, type_id_val))
if resolved:
return resolved
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